--- mqtt/src/daemon.c 2012/01/25 10:34:14 1.1.2.32 +++ mqtt/src/daemon.c 2012/04/15 22:06:10 1.2.2.7 @@ -27,9 +27,13 @@ initSession(int sock, ait_val_t * __restrict v) } else memset(sess, 0, sizeof(struct tagSession)); - TAILQ_INIT(&sess->sess_sndqueue); + pthread_mutex_init(&sess->sess_mtx, NULL); - str = (const char*) cfg_GetAttribute(&cfg, CFG("mqttd"), CFG("retry")); + 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"); if (!str) sess->sess_retry = DEFAULT_RETRY; else @@ -43,6 +47,23 @@ initSession(int sock, ait_val_t * __restrict v) 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; + } else { + mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_CONNECT, cmdCONNECT); + mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_PUBLISH, cmdPUBLISH); + mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_PUBREL, cmdPUBREL); + mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_SUBSCRIBE, cmdSUBSCRIBE); + mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_UNSUBSCRIBE, cmdUNSUBSCRIBE); + mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_PINGREQ, cmdPINGREQ); + mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_DISCONNECT, cmdDISCONNECT); + } + sess->sess_sock = sock; strlcpy(sess->sess_addr, (char*) AIT_GET_STR(v), sizeof sess->sess_addr); io_freeVar(v); @@ -53,6 +74,7 @@ static void finiSession(struct tagSession *sess, int preservSock) { struct tagStore *store; + register int i; ioTRACE(5); @@ -62,16 +84,20 @@ finiSession(struct tagSession *sess, int preservSock) if (call.FiniSessPUB) call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); - while ((store = TAILQ_FIRST(&sess->sess_sndqueue))) { - TAILQ_REMOVE(&sess->sess_sndqueue, store, st_node); + SESS_ELEM_LOCK(sess); + for (i = 0; i < MQTT_QOS_RESERVED; i++) + while ((store = SLIST_FIRST(&sess->sess_txque[i]))) { + SLIST_REMOVE_HEAD(&sess->sess_txque[i], 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); + 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); + } + SESS_ELEM_UNLOCK(sess); + pthread_mutex_destroy(&sess->sess_mtx); if (sess->sess_will.msg) free(sess->sess_will.msg); @@ -81,6 +107,7 @@ finiSession(struct tagSession *sess, int preservSock) if (sess->sess_sock > STDERR_FILENO && !preservSock) srv_Close(sess->sess_sock); + mqtt_srv_Fini(&sess->sess_srv); mqtt_msgFree(&sess->sess_buf, 42); free(sess); @@ -96,9 +123,9 @@ stopSession(struct tagSession *sess) assert(sess); - pthread_mutex_lock(&mtx_sess); + SESS_LOCK; TAILQ_REMOVE(&Sessions, sess, sess_node); - pthread_mutex_unlock(&mtx_sess); + SESS_UNLOCK; ret = mqtt_msgDISCONNECT(&msg); if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) @@ -165,12 +192,10 @@ KASession(struct tagSession *sess) static void * thrSession(struct tagSession *sess) { - mqtt_msg_t msg = { NULL, 0 }; int ret, locKill = 42; struct pollfd pfd; struct mqtthdr *hdr; ait_val_t *v; - struct tagStore *store; pthread_cleanup_push((void(*)(void*)) stopSession, sess); ioTRACE(2); @@ -198,60 +223,30 @@ thrSession(struct tagSession *sess) hdr = (struct mqtthdr*) sess->sess_buf->msg_base; /* dispatch message type */ + if (mqtt_srv_Dispatch(sess->sess_srv, sess)) + ioLIBERR(mqtt); switch (hdr->mqtt_msg.type) { case MQTT_TYPE_CONNECT: ioDEBUG(5, "Exec CONNECT session"); - pthread_mutex_lock(&mtx_sess); - TAILQ_REMOVE(&Sessions, sess, sess_node); - pthread_mutex_unlock(&mtx_sess); - - if (call.FiniSessPUB) - call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); - - 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); - } - - if (sess->sess_will.msg) - free(sess->sess_will.msg); - if (sess->sess_will.topic) - free(sess->sess_will.topic); - if ((v = io_allocVar())) { AIT_SET_STR(v, sess->sess_addr); if (!schedEvent(root, startSession, v, (u_long) sess->sess_sock, sess, ret)) io_freeVar(v); } else ioLIBERR(mqtt); - locKill ^= locKill; - - call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, - sess->sess_addr, sess->sess_user); continue; case MQTT_TYPE_DISCONNECT: ioDEBUG(5, "Exec DISCONNECT session"); - pthread_mutex_lock(&mtx_sess); - TAILQ_REMOVE(&Sessions, sess, sess_node); - pthread_mutex_unlock(&mtx_sess); - finiSession(sess, 0); locKill ^= locKill; - - call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, - sess->sess_addr, sess->sess_user); continue; case MQTT_TYPE_PUBLISH: ioDEBUG(5, "Exec PUBLISH topic QoS=%d", hdr->mqtt_msg.qos); - if (Publish(sess)) + /* + if (cmdPUBLISH(sess)) locKill ^= locKill; + */ break; case MQTT_TYPE_PUBREL: break; @@ -260,6 +255,7 @@ thrSession(struct tagSession *sess) case MQTT_TYPE_UNSUBSCRIBE: break; case MQTT_TYPE_PINGREQ: + ioDEBUG(5, "Exec PINGREQ session"); break; default: ioDEBUG(5, "Error:: Session %s, wrong command %d - DISCARDED", @@ -368,10 +364,10 @@ startSession(sched_task_t *task) pthread_attr_init(&attr); pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); - pthread_mutex_lock(&mtx_sess); + SESS_LOCK; TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); 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, sess->sess_addr, sess->sess_user, sess->sess_ka); @@ -425,6 +421,7 @@ Run(int sock) ioTRACE(1); + /* start alternative thread for scheduler */ if (pthread_create(&tid, NULL, thrSched, NULL)) { ioSYSERR(0); return -1; @@ -446,11 +443,12 @@ Run(int sock) } v = io_allocVar(); if (!v) { - ioLIBERR(mqtt); + ioLIBERR(io); break; } else { memset(str, 0, sizeof str); snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa)); + AIT_FREE_VAL(v); AIT_SET_STR(v, str); } ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v));