--- mqtt/src/mqttd_calls.c 2012/06/01 10:44:51 1.2.2.20 +++ mqtt/src/mqttd_calls.c 2012/06/20 15:01:30 1.2.2.25 @@ -64,24 +64,49 @@ sendPacket(sched_task_t *task) return NULL; } +/* --------------------------------------------------- */ static int -pubOnce(struct tagSession *sess, u_short mid, char * __restrict psTopic, - int topicLen, char * __restrict data, int datlen) +pubOnce(struct tagSession *sess, char * __restrict psTopic, int datlen) { + struct tagPkt *p = NULL; + struct tagSession *s = NULL; + struct tagStore *st = NULL; + regex_t re; + regmatch_t match; + int ret; + char szStr[STRSIZ]; + + TAILQ_FOREACH(s, &Sessions, sess_node) { + SLIST_FOREACH(st, &s->sess_subscr, st_node) { + if ((ret = regcomp(&re, st->st_subscr.sub_topic.msg_base, REG_EXTENDED))) { + regerror(ret, &re, szStr, sizeof szStr); + regfree(&re); + ioDEBUG(3, "Error:: regcomp(%s) %s\n", (char*) + st->st_subscr.sub_topic.msg_base, szStr); + } + 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, s->sess_sock, NULL, 0); + } + + regfree(&re); + } + } + return 0; } static int -pubAck(struct tagSession *sess, u_short mid, char * __restrict psTopic, - int topicLen, char * __restrict data, int datlen) +pubAck(struct tagSession *sess, char * __restrict psTopic, int datlen) { return 0; } static int -pubExactly(struct tagSession *sess, u_short mid, char * __restrict psTopic, - int topicLen, char * __restrict data, int datlen) +pubExactly(struct tagSession *sess, char * __restrict psTopic, int datlen) { return 0; } @@ -92,10 +117,10 @@ cmdPUBLISH(void *srv, int len, void *arg) { struct mqtthdr *hdr; struct tagSession *sess = (struct tagSession*) arg; - void *data = NULL; char szTopic[STRSIZ] = { 0 }; int siz = 0; u_short mid = 0; + struct tagPkt *p = NULL; ioTRACE(2); @@ -103,47 +128,50 @@ cmdPUBLISH(void *srv, int len, void *arg) return -1; ioDEBUG(5, "Exec PUBLISH session"); - siz = mqtt_readPUBLISH(sess->sess_buf, szTopic, sizeof szTopic, &mid, &data); + siz = mqtt_readPUBLISH(sess->sess_buf, szTopic, sizeof szTopic, &mid, NULL); if (siz == -1) { ioDEBUG(5, "Error:: in readPUBLISH #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } + /* duplicate packet for retransmit to subscribers */ + /* + pubpkt = mqtt_msgDup(sess->sess_buf); + if (!pubpkt) { + ioDEBUG(5, "Error:: in duplicate packet #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + return 0; + } else + */ + hdr = (struct mqtthdr*) sess->sess_buf->msg_base; switch (hdr->mqtt_msg.qos) { case MQTT_QOS_ACK: - pubAck(sess, mid, szTopic, sizeof szTopic, data, siz); + pubAck(sess, szTopic, mqtt_pktLen(hdr)); siz = mqtt_msgPUBACK(sess->sess_buf, mid); if (siz == -1) { ioDEBUG(5, "Error:: in msgPUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); - goto end; + return 0; } break; case MQTT_QOS_EXACTLY: - pubExactly(sess, mid, szTopic, sizeof szTopic, data, siz); + pubExactly(sess, szTopic, mqtt_pktLen(hdr)); siz = mqtt_msgPUBREC(sess->sess_buf, mid); if (siz == -1) { ioDEBUG(5, "Error:: in msgPUBREC #%d - %s", mqtt_GetErrno(), mqtt_GetError()); - goto end; + return 0; } break; case MQTT_QOS_ONCE: - pubOnce(sess, mid, szTopic, sizeof szTopic, data, siz); + pubOnce(sess, szTopic, mqtt_pktLen(hdr)); default: - goto end; + return 0; } - 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: - if (data) - free(data); + p = mkPkt(sess->sess_buf->msg_base, siz); + memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); + schedWrite(root, sendPacket, p, sess->sess_sock, NULL, 0); return 0; } @@ -153,6 +181,7 @@ cmdPUBREL(void *srv, int len, void *arg) struct tagSession *sess = (struct tagSession*) arg; int siz = 0; u_short mid = 0; + struct tagPkt *p = NULL; ioTRACE(2); @@ -172,14 +201,12 @@ cmdPUBREL(void *srv, int len, void *arg) if (siz == -1) { ioDEBUG(5, "Error:: in msgPUBCOMP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; - } - if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, MSG_NOSIGNAL)) == -1) - ioSYSERR(0); - else { - ioDEBUG(5, "Sended %d bytes.", siz); + } else { + p = mkPkt(sess->sess_buf->msg_base, siz); memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); } + schedWrite(root, sendPacket, p, sess->sess_sock, NULL, 0); return 0; } @@ -194,6 +221,7 @@ cmdSUBSCRIBE(void *srv, int len, void *arg) struct tagStore *store; char buf[BUFSIZ]; void *ptr; + struct tagPkt *p = NULL; ioTRACE(2); @@ -208,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]); @@ -229,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 */ @@ -259,13 +286,12 @@ cmdSUBSCRIBE(void *srv, int len, void *arg) 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); + } else { + p = mkPkt(sess->sess_buf->msg_base, siz); memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); } + + schedWrite(root, sendPacket, p, sess->sess_sock, NULL, 0); end: mqtt_subFree(&subs); return 0; @@ -280,6 +306,7 @@ cmdUNSUBSCRIBE(void *srv, int len, void *arg) u_short mid = 0; register int i; struct tagStore *store, *tmp; + struct tagPkt *p = NULL; ioTRACE(2); @@ -319,13 +346,12 @@ cmdUNSUBSCRIBE(void *srv, int len, void *arg) 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); + } else { + p = mkPkt(sess->sess_buf->msg_base, siz); memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); } + + schedWrite(root, sendPacket, p, sess->sess_sock, NULL, 0); end: mqtt_subFree(&subs); return 0;