|
|
| version 1.2.2.5, 2012/04/11 15:08:27 | version 1.2.2.6, 2012/04/11 16:37:17 |
|---|---|
| Line 29 initSession(int sock, ait_val_t * __restrict v) | Line 29 initSession(int sock, ait_val_t * __restrict v) |
| pthread_mutex_init(&sess->sess_mtx, NULL); | 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 = cfg_getAttribute(&cfg, "mqttd", "retry"); | str = cfg_getAttribute(&cfg, "mqttd", "retry"); |
| if (!str) | if (!str) |
| Line 72 static void | Line 74 static void |
| finiSession(struct tagSession *sess, int preservSock) | finiSession(struct tagSession *sess, int preservSock) |
| { | { |
| struct tagStore *store; | struct tagStore *store; |
| register int i; | |
| ioTRACE(5); | ioTRACE(5); |
| Line 81 finiSession(struct tagSession *sess, int preservSock) | Line 84 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, "%"); |
| pthread_mutex_lock(&sess->sess_mtx); | SESS_ELEM_LOCK(sess); |
| while ((store = TAILQ_FIRST(&sess->sess_sndqueue))) { | for (i = 0; i < MQTT_QOS_RESERVED; i++) |
| TAILQ_REMOVE(&sess->sess_sndqueue, store, st_node); | 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) | if (store->st_subscr.sub_topic.msg_base) |
| free(store->st_subscr.sub_topic.msg_base); | free(store->st_subscr.sub_topic.msg_base); |
| if (store->st_subscr.sub_value.msg_base) | if (store->st_subscr.sub_value.msg_base) |
| free(store->st_subscr.sub_value.msg_base); | free(store->st_subscr.sub_value.msg_base); |
| free(store); | free(store); |
| } | } |
| pthread_mutex_unlock(&sess->sess_mtx); | SESS_ELEM_UNLOCK(sess); |
| pthread_mutex_destroy(&sess->sess_mtx); | pthread_mutex_destroy(&sess->sess_mtx); |
| if (sess->sess_will.msg) | if (sess->sess_will.msg) |
| Line 119 stopSession(struct tagSession *sess) | Line 123 stopSession(struct tagSession *sess) |
| assert(sess); | assert(sess); |
| pthread_mutex_lock(&mtx_sess); | SESS_LOCK; |
| TAILQ_REMOVE(&Sessions, sess, sess_node); | TAILQ_REMOVE(&Sessions, sess, sess_node); |
| pthread_mutex_unlock(&mtx_sess); | SESS_UNLOCK; |
| ret = mqtt_msgDISCONNECT(&msg); | ret = mqtt_msgDISCONNECT(&msg); |
| if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) | if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) |
| Line 360 startSession(sched_task_t *task) | Line 364 startSession(sched_task_t *task) |
| pthread_attr_init(&attr); | pthread_attr_init(&attr); |
| pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); | pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); |
| pthread_mutex_lock(&mtx_sess); | SESS_LOCK; |
| TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); | TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); |
| pthread_create(&sess->sess_tid, &attr, (void*(*)(void*)) thrSession, sess); | 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, | 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); | sess->sess_addr, sess->sess_user, sess->sess_ka); |