--- mqtt/src/client.c	2011/12/13 08:43:51	1.1
+++ mqtt/src/client.c	2012/01/27 15:05:38	1.2
@@ -0,0 +1,208 @@
+#include "global.h"
+#include "mqtt.h"
+#include "client.h"
+
+
+int
+InitClient(void)
+{
+	int sock;
+
+	sock = socket(args->addr.sa.sa_family, SOCK_STREAM, IPPROTO_TCP);
+	if (sock == -1) {
+		printf("Error:: socket() #%d - %s\n", errno, strerror(errno));
+		return -1;
+	}
+	if (connect(sock, &args->addr.sa, args->addr.sa.sa_len) == -1) {
+		printf("Error:: connect() #%d - %s\n", errno, strerror(errno));
+		return -1;
+	}
+
+	return sock;
+}
+
+int
+ConnectClient(int sock)
+{
+	int siz = 0;
+	struct pollfd pfd;
+
+	siz = mqtt_msgCONNECT(args->msg, (char*) AIT_GET_STR(&args->ConnID), args->ka, 
+			(char*) AIT_GET_STR(&args->User), (char*) AIT_GET_STR(&args->Pass), 
+			(char*) args->Will.Topic.val.string, (char*) args->Will.Msg.val.string, 
+			!args->notClear, args->QoS, args->Retain);
+	if (siz == -1) {
+		printf("Error:: msgCONNECT #%d - %s\n", mqtt_GetErrno(), mqtt_GetError());
+		return -1;
+	}
+
+	siz = send(sock, args->msg->msg_base, siz, 0);
+	if (siz == -1) {
+		printf("Error:: send() #%d - %s\n", errno, strerror(errno));
+		return -1;
+	} else
+		ioVERBOSE(3) printf("Sended CONNECT %d bytes\n", siz);
+
+	pfd.fd = sock;
+	pfd.events = POLLIN | POLLPRI;
+	switch (poll(&pfd, 1, args->ka * 1000)) {
+		case -1:
+			printf("Error:: poll() #%d - %s\n", errno, strerror(errno));
+			return -1;
+		case 0:
+			ioVERBOSE(3) printf("Timeout reached (%d) ...\n", args->ka * 1000);
+			return -1;
+	}
+	if (pfd.revents & (POLLERR | POLLHUP | POLLNVAL))
+		return -1;
+
+	siz = recv(sock, args->msg->msg_base, args->msg->msg_len, 0);
+	if (siz == -1) {
+		printf("Error:: recv() #%d - %s\n", errno, strerror(errno));
+		return -1;
+	} else
+		ioVERBOSE(3) printf("Received %d bytes\n", siz);
+
+	return (u_char) mqtt_readCONNACK(args->msg);
+}
+
+int
+CloseClient(int sock)
+{
+	int siz = 0;
+
+	siz = mqtt_msgDISCONNECT(args->msg);
+	if (siz == -1) {
+		printf("Error:: msgDISCONNECT #%d - %s\n", mqtt_GetErrno(), mqtt_GetError());
+		return -1;
+	}
+
+	siz = send(sock, args->msg->msg_base, siz, 0);
+	if (siz == -1) {
+		printf("Error:: send() #%d - %s\n", errno, strerror(errno));
+		return -1;
+	} else
+		ioVERBOSE(3) printf("Sended DISCONNECT %d bytes\n", siz);
+
+	shutdown(sock, SHUT_RDWR);
+	close(sock);
+	return siz;
+}
+
+int
+KAClient(int sock)
+{
+	int siz = 0;
+	struct pollfd pfd;
+
+	siz = mqtt_msgPINGREQ(args->msg);
+	if (siz == -1) {
+		printf("Error:: msgPINGREQ #%d - %s\n", mqtt_GetErrno(), mqtt_GetError());
+		return -1;
+	}
+	if (SendTo(sock, siz) == -1)
+		return -1;
+
+	pfd.fd = sock;
+	pfd.events = POLLIN | POLLPRI;
+	switch (poll(&pfd, 1, args->ka * 1000)) {
+		case -1:
+			printf("Error:: poll() #%d - %s\n", errno, strerror(errno));
+			return -1;
+		case 0:
+			ioVERBOSE(3) printf("Timeout reached (%d) ...\n", args->ka * 1000);
+			return -1;
+	}
+	if (pfd.revents & (POLLERR | POLLHUP | POLLNVAL))
+		return -1;
+
+	siz = recv(sock, args->msg->msg_base, args->msg->msg_len, 0);
+	if (siz == -1) {
+		printf("Error:: recv() #%d - %s\n", errno, strerror(errno));
+		return -1;
+	} else
+		ioVERBOSE(3) printf("Received %d bytes\n", siz);
+
+	return mqtt_readPINGRESP(args->msg);
+}
+
+void *
+OpenFile(void)
+{
+	int f, siz = 0;
+	void *mem;
+
+	if (!args->isFile)
+		return NULL;
+
+	f = open(AIT_GET_STR(&args->Value), O_RDONLY);
+	if (f == -1) {
+		printf("Error:: in open file #%d - %s\n", errno, strerror(errno));
+		return NULL;
+	}
+	mem = mmap(NULL, siz, PROT_READ, MAP_PRIVATE, f, 0);
+	if (mem == MAP_FAILED) {
+		printf("Error:: in map file #%d - %s\n", errno, strerror(errno));
+		close(f);
+		return NULL;
+	} else
+		close(f);
+
+	AIT_SET_PTR(&args->Value, mem, siz);
+	return mem;
+}
+
+void
+CloseFile(void)
+{
+	if (args->isFile) {
+		munmap(AIT_GET_PTR(&args->Value), AIT_LEN(&args->Value));
+		AIT_FREE_VAL(&args->Value);
+	}
+}
+
+inline int
+SendTo(int sock, int siz)
+{
+	siz = send(sock, args->msg->msg_base, siz, 0);
+	if (siz == -1) {
+		printf("Error:: send() #%d - %s\n", errno, strerror(errno));
+		return -1;
+	} else
+		ioVERBOSE(3) printf("Sended PUBLISH %d bytes\n", siz);
+
+	return siz;
+}
+
+inline int
+RecvFrom(int sock)
+{
+	struct pollfd pfd;
+	int siz = 0;
+
+	pfd.fd = sock;
+	pfd.events = POLLIN | POLLPRI;
+	do {
+		switch (poll(&pfd, 1, args->ka * 1000)) {
+			case -1:
+				printf("Error:: poll() #%d - %s\n", errno, strerror(errno));
+				return -1;
+			case 0:
+				ioVERBOSE(3) printf("Timeout reached (%d) ...\n", args->ka * 1000);
+				if (KAClient(sock) == -1)
+					return -1;
+				continue;
+		}
+		if (pfd.revents & (POLLERR | POLLHUP | POLLNVAL))
+			return -1;
+	} while (0);
+
+	siz = recv(sock, args->msg->msg_base, args->msg->msg_len, 0);
+	if (siz == -1) {
+		printf("Error:: recv() #%d - %s\n", errno, strerror(errno));
+		return -1;
+	} else
+		ioVERBOSE(3) printf("Received %d bytes\n", siz);
+
+	return siz;
+}