version 1.2.2.5, 2012/04/11 15:08:27
|
version 1.2.2.7, 2012/04/15 22:06:10
|
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); |
Line 417 Run(int sock)
|
Line 421 Run(int sock)
|
|
|
ioTRACE(1); |
ioTRACE(1); |
|
|
|
/* start alternative thread for scheduler */ |
if (pthread_create(&tid, NULL, thrSched, NULL)) { |
if (pthread_create(&tid, NULL, thrSched, NULL)) { |
ioSYSERR(0); |
ioSYSERR(0); |
return -1; |
return -1; |
Line 438 Run(int sock)
|
Line 443 Run(int sock)
|
} |
} |
v = io_allocVar(); |
v = io_allocVar(); |
if (!v) { |
if (!v) { |
ioLIBERR(mqtt); | ioLIBERR(io); |
break; |
break; |
} else { |
} else { |
memset(str, 0, sizeof str); |
memset(str, 0, sizeof str); |
snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa)); |
snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa)); |
|
AIT_FREE_VAL(v); |
AIT_SET_STR(v, str); |
AIT_SET_STR(v, str); |
} |
} |
ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v)); |
ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v)); |