--- mqtt/src/daemon.c 2012/01/24 16:28:28 1.1.2.31 +++ mqtt/src/daemon.c 2012/01/25 10:34:14 1.1.2.32 @@ -2,6 +2,7 @@ #include "rtlm.h" #include "utils.h" #include "mqttd.h" +#include "mqttd_calls.h" static void *startSession(sched_task_t *task); @@ -26,6 +27,8 @@ initSession(int sock, ait_val_t * __restrict v) } else memset(sess, 0, sizeof(struct tagSession)); + TAILQ_INIT(&sess->sess_sndqueue); + str = (const char*) cfg_GetAttribute(&cfg, CFG("mqttd"), CFG("retry")); if (!str) sess->sess_retry = DEFAULT_RETRY; @@ -59,8 +62,14 @@ finiSession(struct tagSession *sess, int preservSock) if (call.FiniSessPUB) call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); - while ((store = SLIST_FIRST(&sess->sess_sndqueue))) { - SLIST_REMOVE_HEAD(&sess->sess_sndqueue, st_node); + while ((store = TAILQ_FIRST(&sess->sess_sndqueue))) { + TAILQ_REMOVE(&sess->sess_sndqueue, store, st_node); + + if (store->st_subscr.sub_topic._base) + free(store->st_subscr.sub_topic._base); + if (store->st_subscr.sub_value._base) + free(store->st_subscr.sub_value._base); + free(store); } @@ -199,8 +208,14 @@ thrSession(struct tagSession *sess) if (call.FiniSessPUB) call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); - while ((store = SLIST_FIRST(&sess->sess_sndqueue))) { - SLIST_REMOVE_HEAD(&sess->sess_sndqueue, st_node); + while ((store = TAILQ_FIRST(&sess->sess_sndqueue))) { + TAILQ_REMOVE(&sess->sess_sndqueue, store, st_node); + + if (store->st_subscr.sub_topic._base) + free(store->st_subscr.sub_topic._base); + if (store->st_subscr.sub_value._base) + free(store->st_subscr.sub_value._base); + free(store); } @@ -234,7 +249,9 @@ thrSession(struct tagSession *sess) sess->sess_addr, sess->sess_user); continue; case MQTT_TYPE_PUBLISH: - ioDEBUG(5, "Work in progress ..."); + ioDEBUG(5, "Exec PUBLISH topic QoS=%d", hdr->mqtt_msg.qos); + if (Publish(sess)) + locKill ^= locKill; break; case MQTT_TYPE_PUBREL: break;