--- mqtt/src/mqttd_calls.c 2012/04/11 16:37:17 1.2.2.3 +++ mqtt/src/mqttd_calls.c 2012/05/08 13:04:02 1.2.2.13 @@ -5,7 +5,7 @@ int -cmdPUBLISH(void *srv, void *arg) +cmdPUBLISH(void *srv, int len, void *arg) { struct mqtthdr *hdr; struct tagSession *sess = (struct tagSession*) arg; @@ -33,7 +33,7 @@ cmdPUBLISH(void *srv, void *arg) } int -cmdPUBREL(void *srv, void *arg) +cmdPUBREL(void *srv, int len, void *arg) { struct mqtthdr *hdr; struct tagSession *sess = (struct tagSession*) arg; @@ -49,41 +49,123 @@ cmdPUBREL(void *srv, void *arg) } int -cmdSUBSCRIBE(void *srv, void *arg) +cmdSUBSCRIBE(void *srv, int len, void *arg) { - struct mqtthdr *hdr; struct tagSession *sess = (struct tagSession*) arg; + mqtt_subscr_t *subs = NULL; + int siz = 0; + u_short mid = 0; + register int i; + struct tagStore *store; ioTRACE(2); if (!sess) return -1; - hdr = (struct mqtthdr*) sess->sess_buf->msg_base; + ioDEBUG(5, "Exec SUBSCRIBE session"); + siz = mqtt_readSUBSCRIBE(sess->sess_buf, &mid, &subs); + if (siz == -1) { + ioDEBUG(5, "Error:: in readSUBSCRIBE #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + return 0; + } + /* add to db */ + for (i = 0; i < siz; i++) { + if (call.WritePUB_subscribe(&cfg, pub, mid, subs[i].sub_topic.msg_base, + sess->sess_user, sess->sess_addr, subs[i].sub_ret) > 0) { + store = malloc(sizeof(struct tagStore)); + if (!store) { + ioSYSERR(0); + goto end; + } else { + store->st_msgid = mid; + mqtt_subCopy(&store->st_subscr, &subs[i]); + } + + /* add to cache */ + SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node); + + subs[i].sub_ret = MQTT_QOS_PASS; + } else + subs[i].sub_ret = MQTT_QOS_DENY; + } + + /* send acknowledge */ + siz = mqtt_msgSUBACK(sess->sess_buf, subs, mid); + if (siz == -1) { + ioDEBUG(5, "Error:: in msgSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + goto end; + } + if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, MSG_NOSIGNAL)) == -1) + ioSYSERR(0); + else { + ioDEBUG(5, "Sended %d bytes.", siz); + memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); + } +end: + mqtt_subFree(&subs); return 0; } int -cmdUNSUBSCRIBE(void *srv, void *arg) +cmdUNSUBSCRIBE(void *srv, int len, void *arg) { - struct mqtthdr *hdr; struct tagSession *sess = (struct tagSession*) arg; + mqtt_subscr_t *subs = NULL; + int siz = 0; + u_short mid = 0; + register int i; + struct tagStore *store, *tmp; ioTRACE(2); if (!sess) return -1; - hdr = (struct mqtthdr*) sess->sess_buf->msg_base; + ioDEBUG(5, "Exec UNSUBSCRIBE session"); + siz = mqtt_readUNSUBSCRIBE(sess->sess_buf, &mid, &subs); + if (siz == -1) { + ioDEBUG(5, "Error:: in readUNSUBSCRIBE #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + return 0; + } + /* del from db */ + for (i = 0; i < siz; i++) { + SLIST_FOREACH_SAFE(store, &sess->sess_subscr, st_node, tmp) { + if (store->st_subscr.sub_ret == subs[i].sub_ret && + store->st_subscr.sub_topic.msg_base && + !strcmp(store->st_subscr.sub_topic.msg_base, + subs[i].sub_topic.msg_base)) { + SLIST_REMOVE(&sess->sess_subscr, store, tagStore, st_node); + free(store); + } + } + + call.DeletePUB_subscribe(&cfg, pub, subs[i].sub_topic.msg_base, + sess->sess_user, sess->sess_addr); + } + + /* send acknowledge */ + siz = mqtt_msgUNSUBACK(sess->sess_buf, mid); + if (siz == -1) { + ioDEBUG(5, "Error:: in msgUNSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + goto end; + } + if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, MSG_NOSIGNAL)) == -1) + ioSYSERR(0); + else { + ioDEBUG(5, "Sended %d bytes.", siz); + memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); + } +end: + mqtt_subFree(&subs); return 0; } int -cmdPINGREQ(void *srv, void *arg) +cmdPINGREQ(void *srv, int len, void *arg) { - struct mqtthdr *hdr; struct tagSession *sess = (struct tagSession*) arg; int siz = 0; @@ -92,53 +174,50 @@ cmdPINGREQ(void *srv, void *arg) if (!sess) return -1; - hdr = (struct mqtthdr*) sess->sess_buf->msg_base; + ioDEBUG(5, "Exec PINGREQ session"); siz = mqtt_msgPINGRESP(sess->sess_buf); if (siz == -1) { ioDEBUG(5, "Error:: in msgPINGRESP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } - if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, 0)) == -1) { + if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, MSG_NOSIGNAL)) == -1) { ioSYSERR(0); return 0; - } else + } else { ioDEBUG(5, "Sended %d bytes.", siz); + memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); + } return 0; } int -cmdCONNECT(void *srv, void *arg) +cmdCONNECT(void *srv, int len, void *arg) { struct tagStore *store; struct tagSession *sess = (struct tagSession*) arg; - register int i; ioTRACE(2); if (!sess) return -1; - SESS_LOCK; + ioDEBUG(5, "Exec CONNECT session"); TAILQ_REMOVE(&Sessions, sess, sess_node); - SESS_UNLOCK; if (call.FiniSessPUB) call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); - 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); + while ((store = SLIST_FIRST(&sess->sess_subscr))) { + SLIST_REMOVE_HEAD(&sess->sess_subscr, 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); + 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); - } - SESS_ELEM_UNLOCK(sess); + free(store); + } if (sess->sess_will.msg) free(sess->sess_will.msg); @@ -147,11 +226,12 @@ cmdCONNECT(void *srv, void *arg) call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, sess->sess_addr, sess->sess_user); - return 0; + + return -3; /* reconnect client */ } int -cmdDISCONNECT(void *srv, void *arg) +cmdDISCONNECT(void *srv, int len, void *arg) { struct tagSession *sess = (struct tagSession*) arg; @@ -160,11 +240,10 @@ cmdDISCONNECT(void *srv, void *arg) if (!sess) return -1; - SESS_LOCK; - TAILQ_REMOVE(&Sessions, sess, sess_node); - SESS_UNLOCK; + ioDEBUG(5, "Exec DISCONNECT session"); call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, sess->sess_addr, sess->sess_user); - return 0; + + return -2; /* must terminate dispatcher */ }