--- mqtt/src/mqttd_calls.c 2012/06/20 09:23:28 1.2.2.23 +++ mqtt/src/mqttd_calls.c 2012/06/29 14:44:08 1.2.2.31 @@ -4,55 +4,44 @@ #include "mqttd_calls.h" -static inline struct tagPkt * +static inline ait_val_t * mkPkt(void * __restrict data, int dlen) { - struct tagPkt *p = NULL; + ait_val_t *p = NULL; - p = io_malloc(sizeof(struct tagPkt)); - if (!p) { + if (!(p = io_allocVar())) { ioDEBUG(7, "Error:: in send packet prepare #%d - %s", io_GetErrno(), io_GetError()); return NULL; - } else - memset(p, 0, sizeof(struct tagPkt)); - - p->pkt_data = io_allocVar(); - if (!p->pkt_data) { - ioDEBUG(7, "Error:: in send packet prepare #%d - %s", io_GetErrno(), io_GetError()); - io_free(p); - return NULL; } if (data && dlen > 0) - AIT_SET_BUF(p->pkt_data, data, dlen); + AIT_SET_BUF(p, data, dlen); return p; } static inline void -freePkt(struct tagPkt * __restrict p) +freePkt(ait_val_t ** __restrict p) { if (!p) return; - io_freeVar(&p->pkt_data); - io_free(p); + io_freeVar(p); } static void * sendPacket(sched_task_t *task) { - struct tagPkt *p = TASK_ARG(task); + ait_val_t *p = TASK_ARG(task); register int n, slen; u_char *pos; - if (!p || !p->pkt_data || AIT_ISEMPTY(p->pkt_data)) { + if (!p || AIT_ISEMPTY(p)) { ioDEBUG(9, "Error:: invalid packet or found empty content ..."); return NULL; } - for (slen = AIT_LEN(p->pkt_data), pos = AIT_GET_BUF(p->pkt_data); slen > 0; - slen -= n, pos += n) { + for (slen = AIT_LEN(p), pos = AIT_GET_BUF(p); slen > 0; slen -= n, pos += n) { n = send(TASK_FD(task), pos, slen, MSG_NOSIGNAL); if (n == -1) { ioSYSERR(0); @@ -60,7 +49,7 @@ sendPacket(sched_task_t *task) } } - freePkt(p); + freePkt(&p); return NULL; } @@ -69,16 +58,17 @@ sendPacket(sched_task_t *task) static int pubOnce(struct tagSession *sess, char * __restrict psTopic, int datlen) { - struct tagPkt *p = NULL; + ait_val_t *p = NULL; struct tagSession *s = NULL; - struct tagStore *st = NULL; + struct tagStore *st_, *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) { + SLIST_FOREACH_SAFE(st, &s->sess_subscr, st_node, st_) { if ((ret = regcomp(&re, st->st_subscr.sub_topic.msg_base, REG_EXTENDED))) { regerror(ret, &re, szStr, sizeof szStr); regfree(&re); @@ -87,7 +77,6 @@ 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, s->sess_sock, NULL, 0); } @@ -100,14 +89,100 @@ pubOnce(struct tagSession *sess, char * __restrict psT } static int -pubAck(struct tagSession *sess, char * __restrict psTopic, int datlen) +pubAck(struct tagSession *sess, u_short mid, char * __restrict psTopic, int datlen) { + ait_val_t *p = NULL; + struct tagSession *s = NULL; + struct tagStore *st_, *st = NULL; + regex_t re; + regmatch_t match; + int ret; + char szStr[STRSIZ]; + struct mqtthdr *hdr; + + p = mkPkt(sess->sess_buf->msg_base, datlen); + hdr = (struct mqtthdr*) sess->sess_buf->msg_base; + + /* write topic to database */ + call.DeletePUB_topic(&cfg, pub, sess->sess_cid, mid, psTopic, sess->sess_user, + sess->sess_addr, hdr->mqtt_msg.retain); + call.WritePUB_topic(&cfg, pub, sess->sess_cid, mid, psTopic, AIT_GET_BUF(p), AIT_LEN(p), + sess->sess_user, sess->sess_addr, hdr->mqtt_msg.qos, hdr->mqtt_msg.retain); + + TAILQ_FOREACH(s, &Sessions, sess_node) { + SLIST_FOREACH_SAFE(st, &s->sess_subscr, st_node, st_) { + /* check for QoS */ + if (st->st_subscr.sub_ret < MQTT_QOS_ACK) + continue; + + 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 */ + schedWrite(root, sendPacket, mkPkt(AIT_GET_BUF(p), AIT_LEN(p)), + s->sess_sock, NULL, 0); + } + + regfree(&re); + } + } + + /* delete not retain message */ + call.DeletePUB_topic(&cfg, pub, sess->sess_cid, mid, psTopic, sess->sess_user, + sess->sess_addr, 0); + + freePkt(&p); return 0; } static int -pubExactly(struct tagSession *sess, char * __restrict psTopic, int datlen) +pubExactly(struct tagSession *sess, u_short mid, char * __restrict psTopic, int datlen) { + ait_val_t *p = NULL; + struct tagSession *s = NULL; + struct tagStore *st_, *st = NULL; + regex_t re; + regmatch_t match; + int ret; + char szStr[STRSIZ]; + struct mqtthdr *hdr; + + p = mkPkt(sess->sess_buf->msg_base, datlen); + hdr = (struct mqtthdr*) sess->sess_buf->msg_base; + + /* write topic to database */ + call.DeletePUB_topic(&cfg, pub, sess->sess_cid, mid, psTopic, sess->sess_user, + sess->sess_addr, hdr->mqtt_msg.retain); + call.WritePUB_topic(&cfg, pub, sess->sess_cid, mid, psTopic, AIT_GET_BUF(p), AIT_LEN(p), + sess->sess_user, sess->sess_addr, hdr->mqtt_msg.qos, hdr->mqtt_msg.retain); + + TAILQ_FOREACH(s, &Sessions, sess_node) { + SLIST_FOREACH_SAFE(st, &s->sess_subscr, st_node, st_) { + /* check for QoS */ + if (st->st_subscr.sub_ret < MQTT_QOS_EXACTLY) + continue; + + 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 */ + schedWrite(root, sendPacket, mkPkt(AIT_GET_BUF(p), AIT_LEN(p)), + s->sess_sock, NULL, 0); + } + + regfree(&re); + } + } + + freePkt(&p); return 0; } @@ -120,7 +195,7 @@ cmdPUBLISH(void *srv, int len, void *arg) char szTopic[STRSIZ] = { 0 }; int siz = 0; u_short mid = 0; - struct tagPkt *p = NULL; + ait_val_t *p = NULL; ioTRACE(2); @@ -134,19 +209,11 @@ cmdPUBLISH(void *srv, int len, void *arg) 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, szTopic, mqtt_pktLen(hdr)); + if (pubAck(sess, mid, szTopic, mqtt_pktLen(hdr))) + return 0; siz = mqtt_msgPUBACK(sess->sess_buf, mid); if (siz == -1) { ioDEBUG(5, "Error:: in msgPUBACK #%d - %s", @@ -155,7 +222,8 @@ cmdPUBLISH(void *srv, int len, void *arg) } break; case MQTT_QOS_EXACTLY: - pubExactly(sess, szTopic, mqtt_pktLen(hdr)); + if (pubExactly(sess, mid, szTopic, mqtt_pktLen(hdr))) + return 0; siz = mqtt_msgPUBREC(sess->sess_buf, mid); if (siz == -1) { ioDEBUG(5, "Error:: in msgPUBREC #%d - %s", @@ -181,7 +249,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; + ait_val_t *p = NULL; ioTRACE(2); @@ -195,17 +263,18 @@ cmdPUBREL(void *srv, int len, void *arg) return 0; } - // TODO:: Delete from database topic + /* delete not retain message */ + call.DeletePUB_topic(&cfg, pub, sess->sess_cid, mid, "%", sess->sess_user, + sess->sess_addr, 0); siz = mqtt_msgPUBCOMP(sess->sess_buf, mid); if (siz == -1) { ioDEBUG(5, "Error:: in msgPUBCOMP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; - } else { - p = mkPkt(sess->sess_buf->msg_base, siz); - memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); } + 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; } @@ -221,7 +290,7 @@ cmdSUBSCRIBE(void *srv, int len, void *arg) struct tagStore *store; char buf[BUFSIZ]; void *ptr; - struct tagPkt *p = NULL; + ait_val_t *p = NULL; ioTRACE(2); @@ -236,41 +305,34 @@ cmdSUBSCRIBE(void *srv, int len, void *arg) } /* add to db */ - for (i = 0; i < siz; i++) { - /* 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; + for (i = 0; i < siz; i++, subs[i].sub_ret = MQTT_QOS_DENY) { + store = io_malloc(sizeof(struct tagStore)); + if (!store) { + ioSYSERR(0); + continue; + } else { + store->st_msgid = mid; + mqtt_subCopy(&store->st_subscr, &subs[i]); } - 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; - } else { - store->st_msgid = mid; - mqtt_subCopy(&store->st_subscr, &subs[i]); - } - /* add to cache */ - SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node); + /* add to cache */ + 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) { - ioDEBUG(5, "Error:: in regexp #%d - %s", mqtt_GetErrno(), mqtt_GetError()); - goto end; + /* convert topic to regexp */ + 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()); + + subs[i].sub_ret = MQTT_QOS_DENY; + } else { + ptr = realloc(store->st_subscr.sub_topic.msg_base, strlen(buf) + 1); + if (!ptr) { + ioSYSERR(0); + continue; } else { - ptr = realloc(store->st_subscr.sub_topic.msg_base, strlen(buf) + 1); - if (!ptr) { - ioSYSERR(0); - goto end; - } 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); - } + 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, @@ -278,8 +340,10 @@ cmdSUBSCRIBE(void *srv, int len, void *arg) 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; + } + + call.WritePUB_subscribe(&cfg, pub, sess->sess_cid, mid, buf, + sess->sess_user, sess->sess_addr, subs[i].sub_ret); } /* send acknowledge */ @@ -307,7 +371,7 @@ cmdUNSUBSCRIBE(void *srv, int len, void *arg) u_short mid = 0; register int i; struct tagStore *store, *tmp; - struct tagPkt *p = NULL; + ait_val_t *p = NULL; ioTRACE(2); @@ -363,7 +427,7 @@ cmdPINGREQ(void *srv, int len, void *arg) { struct tagSession *sess = (struct tagSession*) arg; int siz = 0; - struct tagPkt *p = NULL; + ait_val_t *p = NULL; ioTRACE(2); @@ -388,7 +452,6 @@ int cmdCONNECT(void *srv, int len, void *arg) { struct tagStore *store; - struct tagPkt *p; struct tagSession *sess = (struct tagSession*) arg; ioTRACE(2); @@ -417,13 +480,6 @@ cmdCONNECT(void *srv, int len, void *arg) free(store->st_subscr.sub_value.msg_base); io_free(store); - } - - while ((p = SLIST_FIRST(&sess->sess_sndpkt))) { - SLIST_REMOVE_HEAD(&sess->sess_sndpkt, pkt_node); - - io_freeVar(&p->pkt_data); - io_free(p); } if (sess->sess_will.msg)