--- mqtt/src/daemon.c 2012/01/27 16:19:04 1.2.2.2 +++ mqtt/src/daemon.c 2012/01/30 08:29:55 1.2.2.3 @@ -50,6 +50,14 @@ initSession(int sock, ait_val_t * __restrict v) 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; @@ -179,7 +187,6 @@ thrSession(struct tagSession *sess) struct pollfd pfd; struct mqtthdr *hdr; ait_val_t *v; - struct tagStore *store; pthread_cleanup_push((void(*)(void*)) stopSession, sess); ioTRACE(2); @@ -207,60 +214,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.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); - } - - 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;