--- mqtt/src/client.c	2012/01/05 10:01:20	1.1.2.11
+++ mqtt/src/client.c	2012/04/27 16:41:56	1.2.2.4
@@ -1,5 +1,6 @@
 #include "global.h"
 #include "mqtt.h"
+#include "client.h"
 
 
 int
@@ -55,6 +56,7 @@ ConnectClient(int sock)
 	if (pfd.revents & (POLLERR | POLLHUP | POLLNVAL))
 		return -1;
 
+	memset(args->msg->msg_base, 0, args->msg->msg_len);
 	siz = recv(sock, args->msg->msg_base, args->msg->msg_len, 0);
 	if (siz == -1) {
 		printf("Error:: recv() #%d - %s\n", errno, strerror(errno));
@@ -76,7 +78,7 @@ CloseClient(int sock)
 		return -1;
 	}
 
-	siz = send(sock, args->msg->msg_base, siz, 0);
+	siz = send(sock, args->msg->msg_base, siz, MSG_NOSIGNAL);
 	if (siz == -1) {
 		printf("Error:: send() #%d - %s\n", errno, strerror(errno));
 		return -1;
@@ -88,48 +90,6 @@ CloseClient(int 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;
-	}
-
-	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 PINGREQ %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_readPINGRESP(args->msg);
-}
-
 void *
 OpenFile(void)
 {
@@ -163,4 +123,52 @@ CloseFile(void)
 		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, MSG_NOSIGNAL);
+	if (siz == -1) {
+		printf("Error:: send() #%d - %s\n", errno, strerror(errno));
+		return -1;
+	} else
+		ioVERBOSE(3) printf("Sended %d bytes\n", siz);
+
+	return siz;
+}
+
+inline int
+RecvFrom(int sock)
+{
+	struct pollfd pfd;
+	int siz = 0;
+
+	memset(args->msg->msg_base, 0, args->msg->msg_len);
+
+	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 (mqtt_KeepAlive(sock, args->ka, 1) == -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;
 }