--- mqtt/src/daemon.c	2012/04/25 13:33:53	1.2.2.13
+++ mqtt/src/daemon.c	2012/04/27 10:07:50	1.2.2.19
@@ -23,7 +23,6 @@ initSession(int sock, ait_val_t * __restrict v)
 	sess = malloc(sizeof(struct tagSession));
 	if (!sess) {
 		ioSYSERR(0);
-		io_freeVar(v);
 		return NULL;
 	} else
 		memset(sess, 0, sizeof(struct tagSession));
@@ -38,19 +37,10 @@ initSession(int sock, ait_val_t * __restrict v)
 	else
 		sess->sess_retry = strtol(str, NULL, 0);
 
-	if (!(sess->sess_root = schedBegin())) {
-		ioLIBERR(sched);
-		free(sess);
-		io_freeVar(v);
-		return NULL;
-	}
-
 	sess->sess_buf = mqtt_msgAlloc(USHRT_MAX);
 	if (!sess->sess_buf) {
 		ioLIBERR(mqtt);
-		schedEnd(&sess->sess_root);
 		free(sess);
-		io_freeVar(v);
 		return NULL;
 	}
 
@@ -59,9 +49,7 @@ initSession(int sock, ait_val_t * __restrict v)
 	if (!sess->sess_srv) {
 		ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError());
 		mqtt_msgFree(&sess->sess_buf, 42);
-		schedEnd(&sess->sess_root);
 		free(sess);
-		io_freeVar(v);
 		return NULL;
 	} else {
 		mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_CONNECT, cmdCONNECT);
@@ -75,7 +63,6 @@ initSession(int sock, ait_val_t * __restrict v)
 
 	sess->sess_sock = sock;
 	strlcpy(sess->sess_addr, (char*) AIT_GET_STR(v), sizeof sess->sess_addr);
-	io_freeVar(v);
 	return sess;
 }
 
@@ -93,7 +80,6 @@ finiSession(struct tagSession *sess)
 		call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");
 
 	SESS_ELEM_LOCK(sess);
-
 	while ((store = SLIST_FIRST(&sess->sess_subscr))) {
 		SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node);
 
@@ -107,8 +93,6 @@ finiSession(struct tagSession *sess)
 	SESS_ELEM_UNLOCK(sess);
 	pthread_mutex_destroy(&sess->sess_mtx);
 
-	schedEnd(&sess->sess_root);
-
 	if (sess->sess_will.msg)
 		free(sess->sess_will.msg);
 	if (sess->sess_will.topic)
@@ -148,59 +132,12 @@ stopSession(struct tagSession *sess)
 			sess->sess_addr, sess->sess_user);
 }
 
-static int
-KASession(struct tagSession *sess)
-{
-	mqtt_msg_t msg = { NULL, 0 };
-	int ret;
-	struct pollfd pfd;
-
-	ioTRACE(4);
-
-	assert(sess);
-
-	/* ping request */
-	ret = mqtt_msgPINGREQ(&msg);
-	if ((ret = send(sess->sess_sock, msg.msg_base, ret, MSG_NOSIGNAL)) == -1) {
-		ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
-		return -1;
-	} else {
-		ioDEBUG(5, "Sended %d bytes for ping request", ret);
-		free(msg.msg_base);
-		memset(&msg, 0, sizeof msg);
-	}
-
-	pfd.fd = sess->sess_sock;
-	pfd.events = POLLIN | POLLPRI;
-	if ((ret = poll(&pfd, 1, sess->sess_ka * 1000)) == -1 || 
-			pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
-		ioDEBUG(3, "Error:: poll(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
-		return -1;
-	} else if (!ret) {
-		ioDEBUG(5, "Warning:: Session is abandoned ... must be disconnect!");
-		return 1;
-	}
-	/* receive & decode packet */
-	if (recv(sess->sess_sock, sess->sess_buf->msg_base, sess->sess_buf->msg_len, 0) == -1) {
-		ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
-		return -1;
-	}
-	if (mqtt_readPINGRESP(sess->sess_buf)) {
-		ioDEBUG(5, "Warning:: Session is broken, not hear ping response ... must be disconnect!");
-		return 2;
-	}
-
-	/* Keep Alive is OK! */
-	return 0;
-}
-
 static void *
 thrSession(struct tagSession *sess)
 {
 	int ret, locKill = 42;
 	struct pollfd pfd;
 	struct mqtthdr *hdr;
-	ait_val_t *v;
 
 	pthread_cleanup_push((void(*)(void*)) stopSession, sess);
 	ioTRACE(2);
@@ -212,7 +149,7 @@ thrSession(struct tagSession *sess)
 				pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
 			ioDEBUG(3, "Error:: poll(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
 			break;
-		} else if (!ret && (ret = KASession(sess))) {
+		} else if (!ret && (ret = mqtt_KeepAlive(sess->sess_sock, sess->sess_ka, 1))) {
 			call.LOG(logg, "Session %s keep-alive missing from %s for user %s ...\n", 
 					sess->sess_cid, sess->sess_addr, sess->sess_user);
 			break;
@@ -232,30 +169,16 @@ thrSession(struct tagSession *sess)
 			ioLIBERR(mqtt);
 		switch (hdr->mqtt_msg.type) {
 			case MQTT_TYPE_CONNECT:
-				ioDEBUG(5, "Exec CONNECT session");
-				if ((v = io_allocVar())) {
-					AIT_SET_STR(v, sess->sess_addr);
-					if (!schedEvent(root, startSession, v, (u_long) sess->sess_sock, sess, ret))
-						io_freeVar(v);
-				} else
-					ioLIBERR(mqtt);
-
-				SESS_LOCK;
-				TAILQ_REMOVE(&Sessions, sess, sess_node);
-				SESS_UNLOCK;
-
+				schedEvent(root, startSession, NULL, (u_long) sess->sess_sock, sess, ret);
 				locKill ^= locKill;
 				break;
 			case MQTT_TYPE_DISCONNECT:
-				ioDEBUG(5, "Exec DISCONNECT session");
 				finiSession(sess);
-
-				SESS_LOCK;
-				TAILQ_REMOVE(&Sessions, sess, sess_node);
-				SESS_UNLOCK;
-
 				locKill ^= locKill;
 				continue;
+			case MQTT_TYPE_SUBSCRIBE:
+			case MQTT_TYPE_PINGREQ:
+				break;
 			case MQTT_TYPE_PUBLISH:
 				ioDEBUG(5, "Exec PUBLISH topic QoS=%d", hdr->mqtt_msg.qos);
 				/*
@@ -265,16 +188,8 @@ thrSession(struct tagSession *sess)
 				break;
 			case MQTT_TYPE_PUBREL:
 				break;
-			case MQTT_TYPE_SUBSCRIBE:
-				break;
 			case MQTT_TYPE_UNSUBSCRIBE:
 				break;
-			case MQTT_TYPE_PINGREQ:
-				ioDEBUG(5, "Exec PINGREQ session");
-				break;
-			case MQTT_TYPE_PINGRESP:
-				ioDEBUG(5, "Exec PINGRESP session");
-				break;
 			default:
 				ioDEBUG(5, "Error:: Session %s, wrong command %d - DISCARDED", 
 						sess->sess_cid, hdr->mqtt_msg.type);
@@ -304,8 +219,8 @@ startSession(sched_task_t *task)
 	if (!TASK_DATA(task)) {
 		/* flow from accept new clients */
 		sess = initSession(TASK_FD(task), TASK_ARG(task));
+		io_freeVar(TASK_ARG(task));
 		if (!sess) {
-			io_freeVar(TASK_ARG(task));
 			close(TASK_FD(task));
 			return NULL;
 		}
@@ -431,6 +346,7 @@ acceptClient(sched_task_t *task)
 	} else {
 		memset(str, 0, sizeof str);
 		snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa));
+		AIT_FREE_VAL(v);
 		AIT_SET_STR(v, str);
 	}
 	ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v));