|
|
| version 1.1.2.31, 2012/01/24 16:28:28 | version 1.2.2.2, 2012/01/27 16:19:04 |
|---|---|
| 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 40 initSession(int sock, ait_val_t * __restrict v) | Line 43 initSession(int sock, ait_val_t * __restrict v) |
| return NULL; | return NULL; |
| } | } |
| 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()); | |
| mqtt_msgFree(&sess->sess_buf, 42); | |
| free(sess); | |
| io_freeVar(v); | |
| return NULL; | |
| } | |
| sess->sess_sock = sock; | sess->sess_sock = sock; |
| strlcpy(sess->sess_addr, (char*) AIT_GET_STR(v), sizeof sess->sess_addr); | strlcpy(sess->sess_addr, (char*) AIT_GET_STR(v), sizeof sess->sess_addr); |
| io_freeVar(v); | io_freeVar(v); |
| Line 59 finiSession(struct tagSession *sess, int preservSock) | Line 71 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.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); |
| } | } |
| Line 72 finiSession(struct tagSession *sess, int preservSock) | Line 90 finiSession(struct tagSession *sess, int preservSock) |
| if (sess->sess_sock > STDERR_FILENO && !preservSock) | if (sess->sess_sock > STDERR_FILENO && !preservSock) |
| srv_Close(sess->sess_sock); | srv_Close(sess->sess_sock); |
| mqtt_srv_Fini(&sess->sess_srv); | |
| mqtt_msgFree(&sess->sess_buf, 42); | mqtt_msgFree(&sess->sess_buf, 42); |
| free(sess); | free(sess); |
| Line 156 KASession(struct tagSession *sess) | Line 175 KASession(struct tagSession *sess) |
| static void * | static void * |
| thrSession(struct tagSession *sess) | thrSession(struct tagSession *sess) |
| { | { |
| mqtt_msg_t msg = { NULL, 0 }; | |
| int ret, locKill = 42; | int ret, locKill = 42; |
| struct pollfd pfd; | struct pollfd pfd; |
| struct mqtthdr *hdr; | struct mqtthdr *hdr; |
| Line 199 thrSession(struct tagSession *sess) | Line 217 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.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); |
| } | } |
| Line 234 thrSession(struct tagSession *sess) | Line 258 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; |