--- mqtt/src/daemon.c	2012/01/30 08:29:55	1.2.2.3
+++ mqtt/src/daemon.c	2012/04/15 22:06:10	1.2.2.7
@@ -27,9 +27,13 @@ initSession(int sock, ait_val_t * __restrict v)
 	} else
 		memset(sess, 0, sizeof(struct tagSession));
 
-	TAILQ_INIT(&sess->sess_sndqueue);
+	pthread_mutex_init(&sess->sess_mtx, NULL);
 
-	str = (const char*) cfg_GetAttribute(&cfg, CFG("mqttd"), CFG("retry"));
+	SLIST_INIT(&sess->sess_txque[MQTT_QOS_ONCE]);
+	SLIST_INIT(&sess->sess_txque[MQTT_QOS_ACK]);
+	SLIST_INIT(&sess->sess_txque[MQTT_QOS_EXACTLY]);
+
+	str = cfg_getAttribute(&cfg, "mqttd", "retry");
 	if (!str)
 		sess->sess_retry = DEFAULT_RETRY;
 	else
@@ -70,6 +74,7 @@ static void
 finiSession(struct tagSession *sess, int preservSock)
 {
 	struct tagStore *store;
+	register int i;
 
 	ioTRACE(5);
 
@@ -79,16 +84,20 @@ finiSession(struct tagSession *sess, int preservSock)
 	if (call.FiniSessPUB)
 		call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");
 
-	while ((store = TAILQ_FIRST(&sess->sess_sndqueue))) {
-		TAILQ_REMOVE(&sess->sess_sndqueue, store, st_node);
+	SESS_ELEM_LOCK(sess);
+	for (i = 0; i < MQTT_QOS_RESERVED; i++)
+		while ((store = SLIST_FIRST(&sess->sess_txque[i]))) {
+			SLIST_REMOVE_HEAD(&sess->sess_txque[i], st_node);
 
-		if (store->st_subscr.sub_topic.msg_base)
-			free(store->st_subscr.sub_topic.msg_base);
-		if (store->st_subscr.sub_value.msg_base)
-			free(store->st_subscr.sub_value.msg_base);
+			if (store->st_subscr.sub_topic.msg_base)
+				free(store->st_subscr.sub_topic.msg_base);
+			if (store->st_subscr.sub_value.msg_base)
+				free(store->st_subscr.sub_value.msg_base);
 
-		free(store);
-	}
+			free(store);
+		}
+	SESS_ELEM_UNLOCK(sess);
+	pthread_mutex_destroy(&sess->sess_mtx);
 
 	if (sess->sess_will.msg)
 		free(sess->sess_will.msg);
@@ -114,9 +123,9 @@ stopSession(struct tagSession *sess)
 
 	assert(sess);
 
-	pthread_mutex_lock(&mtx_sess);
+	SESS_LOCK;
 	TAILQ_REMOVE(&Sessions, sess, sess_node);
-	pthread_mutex_unlock(&mtx_sess);
+	SESS_UNLOCK;
 
 	ret = mqtt_msgDISCONNECT(&msg);
 	if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1)
@@ -246,6 +255,7 @@ thrSession(struct tagSession *sess)
 			case MQTT_TYPE_UNSUBSCRIBE:
 				break;
 			case MQTT_TYPE_PINGREQ:
+				ioDEBUG(5, "Exec PINGREQ session");
 				break;
 			default:
 				ioDEBUG(5, "Error:: Session %s, wrong command %d - DISCARDED", 
@@ -354,10 +364,10 @@ startSession(sched_task_t *task)
 	pthread_attr_init(&attr);
 	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
 
-	pthread_mutex_lock(&mtx_sess);
+	SESS_LOCK;
 	TAILQ_INSERT_TAIL(&Sessions, sess, sess_node);
 	pthread_create(&sess->sess_tid, &attr, (void*(*)(void*)) thrSession, sess);
-	pthread_mutex_unlock(&mtx_sess);
+	SESS_UNLOCK;
 
 	call.LOG(logg, "Session %s started from %s for user %s (timeout=%d) OK!\n", sess->sess_cid, 
 			sess->sess_addr, sess->sess_user, sess->sess_ka);
@@ -411,6 +421,7 @@ Run(int sock)
 
 	ioTRACE(1);
 
+	/* start alternative thread for scheduler */
 	if (pthread_create(&tid, NULL, thrSched, NULL)) {
 		ioSYSERR(0);
 		return -1;
@@ -432,11 +443,12 @@ Run(int sock)
 		}
 		v = io_allocVar();
 		if (!v) {
-			ioLIBERR(mqtt);
+			ioLIBERR(io);
 			break;
 		} 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));