version 1.1.2.31, 2012/01/24 16:28:28
|
version 1.1.2.32, 2012/01/25 10:34:14
|
Line 2
|
Line 2
|
#include "rtlm.h" |
#include "rtlm.h" |
#include "utils.h" |
#include "utils.h" |
#include "mqttd.h" |
#include "mqttd.h" |
|
#include "mqttd_calls.h" |
|
|
|
|
static void *startSession(sched_task_t *task); |
static void *startSession(sched_task_t *task); |
Line 26 initSession(int sock, ait_val_t * __restrict v)
|
Line 27 initSession(int sock, ait_val_t * __restrict v)
|
} else |
} else |
memset(sess, 0, sizeof(struct tagSession)); |
memset(sess, 0, sizeof(struct tagSession)); |
|
|
|
TAILQ_INIT(&sess->sess_sndqueue); |
|
|
str = (const char*) cfg_GetAttribute(&cfg, CFG("mqttd"), CFG("retry")); |
str = (const char*) cfg_GetAttribute(&cfg, CFG("mqttd"), CFG("retry")); |
if (!str) |
if (!str) |
sess->sess_retry = DEFAULT_RETRY; |
sess->sess_retry = DEFAULT_RETRY; |
Line 59 finiSession(struct tagSession *sess, int preservSock)
|
Line 62 finiSession(struct tagSession *sess, int preservSock)
|
if (call.FiniSessPUB) |
if (call.FiniSessPUB) |
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); |
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); |
|
|
while ((store = SLIST_FIRST(&sess->sess_sndqueue))) { | while ((store = TAILQ_FIRST(&sess->sess_sndqueue))) { |
SLIST_REMOVE_HEAD(&sess->sess_sndqueue, st_node); | 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); |
free(store); |
} |
} |
|
|
Line 199 thrSession(struct tagSession *sess)
|
Line 208 thrSession(struct tagSession *sess)
|
if (call.FiniSessPUB) |
if (call.FiniSessPUB) |
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); |
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); |
|
|
while ((store = SLIST_FIRST(&sess->sess_sndqueue))) { | while ((store = TAILQ_FIRST(&sess->sess_sndqueue))) { |
SLIST_REMOVE_HEAD(&sess->sess_sndqueue, st_node); | 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); |
free(store); |
} |
} |
|
|
Line 234 thrSession(struct tagSession *sess)
|
Line 249 thrSession(struct tagSession *sess)
|
sess->sess_addr, sess->sess_user); |
sess->sess_addr, sess->sess_user); |
continue; |
continue; |
case MQTT_TYPE_PUBLISH: |
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; |
break; |
case MQTT_TYPE_PUBREL: |
case MQTT_TYPE_PUBREL: |
break; |
break; |