--- mqtt/src/daemon.c	2012/01/30 13:39:22	1.2.2.4
+++ mqtt/src/daemon.c	2012/04/15 23:45:34	1.2.2.8
@@ -29,9 +29,11 @@ initSession(int sock, ait_val_t * __restrict v)
 
 	pthread_mutex_init(&sess->sess_mtx, NULL);
 
-	TAILQ_INIT(&sess->sess_sndqueue);
+	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 = (const char*) cfg_GetAttribute(&cfg, CFG("mqttd"), CFG("retry"));
+	str = cfg_getAttribute(&cfg, "mqttd", "retry");
 	if (!str)
 		sess->sess_retry = DEFAULT_RETRY;
 	else
@@ -45,6 +47,7 @@ initSession(int sock, ait_val_t * __restrict v)
 		return NULL;
 	}
 
+	/* init server actor */
 	sess->sess_srv = mqtt_srv_Init(sock, sess->sess_buf);
 	if (!sess->sess_srv) {
 		ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError());
@@ -72,6 +75,7 @@ static void
 finiSession(struct tagSession *sess, int preservSock)
 {
 	struct tagStore *store;
+	register int i;
 
 	ioTRACE(5);
 
@@ -81,18 +85,19 @@ finiSession(struct tagSession *sess, int preservSock)
 	if (call.FiniSessPUB)
 		call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");
 
-	pthread_mutex_lock(&sess->sess_mtx);
-	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);
-	}
-	pthread_mutex_unlock(&sess->sess_mtx);
+			free(store);
+		}
+	SESS_ELEM_UNLOCK(sess);
 	pthread_mutex_destroy(&sess->sess_mtx);
 
 	if (sess->sess_will.msg)
@@ -119,9 +124,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)
@@ -280,6 +285,7 @@ startSession(sched_task_t *task)
 	assert(task);
 
 	if (!TASK_DATA(task)) {
+		/* flow from accept new clients */
 		sess = initSession(TASK_FD(task), TASK_ARG(task));
 		if (!sess) {
 			io_freeVar(TASK_ARG(task));
@@ -360,10 +366,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);
@@ -391,10 +397,11 @@ static void *
 thrSched(void *arg __unused)
 {
 	struct tagSession *sess;
-	struct timespec pl = { 0, 10000000 };
+	struct timespec pl = { 0, 100000000 };
 
 	ioTRACE(1);
 
+	/* start scheduler loop */
 	schedPolling(root, &pl, NULL);
 	schedRun(root, &Kill);
 
@@ -417,6 +424,7 @@ Run(int sock)
 
 	ioTRACE(1);
 
+	/* start alternative thread for scheduler */
 	if (pthread_create(&tid, NULL, thrSched, NULL)) {
 		ioSYSERR(0);
 		return -1;
@@ -429,6 +437,7 @@ Run(int sock)
 		return -1;
 	}
 
+	/* state machine - accept new connections */
 	while (!Kill) {
 		if ((cli = accept(sock, &sa.sa, &sslen)) == -1) {
 			if (!Kill)
@@ -438,11 +447,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));