--- mqtt/src/mqttd_calls.c 2012/05/05 15:16:34 1.2.2.12 +++ mqtt/src/mqttd_calls.c 2012/05/08 13:04:02 1.2.2.13 @@ -97,7 +97,7 @@ cmdSUBSCRIBE(void *srv, int len, void *arg) 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, 0)) == -1) + if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, MSG_NOSIGNAL)) == -1) ioSYSERR(0); else { ioDEBUG(5, "Sended %d bytes.", siz); @@ -112,12 +112,54 @@ int cmdUNSUBSCRIBE(void *srv, int len, void *arg) { 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; + 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; } @@ -138,7 +180,7 @@ cmdPINGREQ(void *srv, int len, void *arg) 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 {