--- mqtt/src/daemon.c	2012/04/25 12:04:30	1.2.2.11
+++ mqtt/src/daemon.c	2012/05/03 14:00:33	1.2.2.21
@@ -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)
@@ -142,56 +126,10 @@ stopSession(struct tagSession *sess)
 	free(msg.msg_base);
 
 	ioDEBUG(1, "Close socket=%d", sess->sess_sock);
-	finiSession(sess);
-
 	call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, 
 			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;
+	finiSession(sess);
 }
 
 static void *
@@ -200,7 +138,6 @@ 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;
@@ -230,22 +167,18 @@ thrSession(struct tagSession *sess)
 		/* dispatch message type */
 		if (mqtt_srv_Dispatch(sess->sess_srv, sess))
 			ioLIBERR(mqtt);
-		locKill ^= locKill;
 		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);
-				continue;
+				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);
-				locKill = 42;
+				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);
 				/*
@@ -255,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);
@@ -290,14 +215,12 @@ startSession(sched_task_t *task)
 	ioTRACE(4);
 
 	assert(task);
-	printf("aaaaaaaaaaaaaaaaa\n");
-	fflush(stdout);
 
 	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;
 		}
@@ -329,8 +252,6 @@ startSession(sched_task_t *task)
 		sess->sess_will.flag = flg.will_flg;
 	}
 
-	printf("sql=%p\n", acc);
-	fflush(stdout);
 	/* check online table for user */
 	if (call.LoginACC(&cfg, acc, sess->sess_user, sess->sess_pass) < 1) {
 		ioDEBUG(0, "Login:: DENIED for username %s and password %s", 
@@ -341,8 +262,7 @@ startSession(sched_task_t *task)
 		ioDEBUG(1, "Login:: ALLOWED for username %s ...", sess->sess_user);
 		ret = MQTT_RETCODE_ACCEPTED;
 	}
-	printf(".sql=%p\n", pub);
-	fflush(stdout);
+
 	if (call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%") > 0) {
 		ioDEBUG(2, "Old session %s should be disconnect!", sess->sess_cid);
 		TAILQ_FOREACH(s, &Sessions, sess_node)
@@ -352,8 +272,7 @@ startSession(sched_task_t *task)
 				break;
 			}
 	}
-	printf("...sql=%p\n", pub);
-	fflush(stdout);
+
 	if (call.InitSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, sess->sess_addr, 
 				sess->sess_will.flag, sess->sess_will.topic, sess->sess_will.msg, 
 				sess->sess_will.qos, sess->sess_will.retain) == -1) {
@@ -380,8 +299,7 @@ startSession(sched_task_t *task)
 	/* Start session thread OK ... */
 	SESS_LOCK;
 	TAILQ_INSERT_TAIL(&Sessions, sess, sess_node);
-	pthread_create(&sess->sess_tid, NULL, (void*(*)(void*)) thrSession, sess);
-	pthread_detach(sess->sess_tid);
+	pthread_create(&sess->sess_tid, &attr, (void*(*)(void*)) thrSession, sess);
 	SESS_UNLOCK;
 
 	call.LOG(logg, "Session %s started from %s for user %s (timeout=%d) OK!\n", sess->sess_cid, 
@@ -418,7 +336,7 @@ acceptClient(sched_task_t *task)
 	if ((cli = accept(TASK_FD(task), &sa.sa, &sslen)) == -1)
 		goto end;
 	else
-		fcntl(TASK_FD(task), F_SETFL, fcntl(TASK_FD(task), F_GETFL, 0) | O_NONBLOCK);
+		fcntl(cli, F_SETFL, fcntl(cli, F_GETFL, 0) | O_NONBLOCK);
 
 	v = io_allocVar();
 	if (!v) {
@@ -428,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));