--- mqtt/src/mqttd_calls.c 2012/06/19 08:32:17 1.2.2.22 +++ mqtt/src/mqttd_calls.c 2012/06/20 15:01:30 1.2.2.25 @@ -87,8 +87,9 @@ pubOnce(struct tagSession *sess, char * __restrict psT } if (!regexec(&re, psTopic, 1, &match, 0)) { /* MATCH */ + ioDEBUG(1, "+++ dlen=%d\n", datlen); p = mkPkt(sess->sess_buf->msg_base, datlen); - schedWrite(root, sendPacket, p, sess->sess_sock, NULL, 0); + schedWrite(root, sendPacket, p, s->sess_sock, NULL, 0); } regfree(&re); @@ -145,7 +146,7 @@ cmdPUBLISH(void *srv, int len, void *arg) hdr = (struct mqtthdr*) sess->sess_buf->msg_base; switch (hdr->mqtt_msg.qos) { case MQTT_QOS_ACK: - pubAck(sess, szTopic, siz); + pubAck(sess, szTopic, mqtt_pktLen(hdr)); siz = mqtt_msgPUBACK(sess->sess_buf, mid); if (siz == -1) { ioDEBUG(5, "Error:: in msgPUBACK #%d - %s", @@ -154,7 +155,7 @@ cmdPUBLISH(void *srv, int len, void *arg) } break; case MQTT_QOS_EXACTLY: - pubExactly(sess, szTopic, siz); + pubExactly(sess, szTopic, mqtt_pktLen(hdr)); siz = mqtt_msgPUBREC(sess->sess_buf, mid); if (siz == -1) { ioDEBUG(5, "Error:: in msgPUBREC #%d - %s", @@ -163,7 +164,7 @@ cmdPUBLISH(void *srv, int len, void *arg) } break; case MQTT_QOS_ONCE: - pubOnce(sess, szTopic, siz); + pubOnce(sess, szTopic, mqtt_pktLen(hdr)); default: return 0; } @@ -235,18 +236,18 @@ cmdSUBSCRIBE(void *srv, int len, void *arg) } /* add to db */ - for (i = 0; i < siz; i++) { + for (i = 0; i < siz; i++, subs[i].sub_ret = MQTT_QOS_DENY) { /* convert topic to sql search statement */ if (mqtt_sqlTopic(subs[i].sub_topic.msg_base, buf, sizeof buf) == -1) { ioDEBUG(5, "Error:: in db #%d - %s", mqtt_GetErrno(), mqtt_GetError()); - goto end; + continue; } if (call.WritePUB_subscribe(&cfg, pub, sess->sess_cid, mid, buf, sess->sess_user, sess->sess_addr, subs[i].sub_ret) > 0) { store = io_malloc(sizeof(struct tagStore)); if (!store) { ioSYSERR(0); - goto end; + continue; } else { store->st_msgid = mid; mqtt_subCopy(&store->st_subscr, &subs[i]); @@ -256,29 +257,28 @@ cmdSUBSCRIBE(void *srv, int len, void *arg) SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node); /* convert topic to regexp */ - if (mqtt_expandTopic(subs[i].sub_topic.msg_base, buf, sizeof buf, 1, 0) == -1) { + if (mqtt_expandTopic(subs[i].sub_topic.msg_base, buf, sizeof buf, 1, 1) == -1) { ioDEBUG(5, "Error:: in regexp #%d - %s", mqtt_GetErrno(), mqtt_GetError()); - goto end; + + subs[i].sub_ret = MQTT_QOS_DENY; } else { ptr = realloc(store->st_subscr.sub_topic.msg_base, strlen(buf) + 1); - if (!ptr) { + if (!ptr) ioSYSERR(0); - goto end; - } else { + else { store->st_subscr.sub_topic.msg_base = ptr; store->st_subscr.sub_topic.msg_len = strlen(buf) + 1; memcpy(store->st_subscr.sub_topic.msg_base, buf, store->st_subscr.sub_topic.msg_len); } - } - call.LOG(logg, "Added [%s] SUBSCRIBE '%s'(%d) from %s\n", sess->sess_cid, - store->st_subscr.sub_topic.msg_base, - store->st_subscr.sub_topic.msg_len, sess->sess_addr); + call.LOG(logg, "Added [%s] SUBSCRIBE '%s'(%d) from %s\n", sess->sess_cid, + store->st_subscr.sub_topic.msg_base, + store->st_subscr.sub_topic.msg_len, sess->sess_addr); - subs[i].sub_ret = MQTT_QOS_PASS; - } else - subs[i].sub_ret = MQTT_QOS_DENY; + subs[i].sub_ret = MQTT_QOS_PASS; + } + } } /* send acknowledge */