--- mqtt/src/mqttd_calls.c 2012/06/01 10:44:51 1.2.2.20 +++ mqtt/src/mqttd_calls.c 2012/06/29 15:43:13 1.2.2.32 @@ -1,58 +1,48 @@ #include "global.h" #include "mqttd.h" +#include "utils.h" #include "rtlm.h" #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,29 +50,140 @@ sendPacket(sched_task_t *task) } } - freePkt(p); + freePkt(&p); 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) { + 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]; + + + TAILQ_FOREACH(s, &Sessions, sess_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); + ioDEBUG(3, "Error:: regcomp(%s) %s\n", (char*) + st->st_subscr.sub_topic.msg_base, szStr); + } + if (!regexec(&re, psTopic, 1, &match, 0)) { + /* MATCH */ + 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, 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, u_short mid, char * __restrict psTopic, - int topicLen, char * __restrict data, 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; } @@ -92,10 +193,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; + ait_val_t *p = NULL; ioTRACE(2); @@ -103,7 +204,7 @@ 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; @@ -112,38 +213,34 @@ 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, mid, szTopic, sizeof szTopic, data, siz); + 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", mqtt_GetErrno(), mqtt_GetError()); - goto end; + return 0; } break; case MQTT_QOS_EXACTLY: - pubExactly(sess, mid, szTopic, sizeof szTopic, data, siz); + 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", 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 +250,7 @@ cmdPUBREL(void *srv, int len, void *arg) struct tagSession *sess = (struct tagSession*) arg; int siz = 0; u_short mid = 0; + ait_val_t *p = NULL; ioTRACE(2); @@ -166,20 +264,19 @@ 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; } - 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); - } + 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 +291,7 @@ cmdSUBSCRIBE(void *srv, int len, void *arg) struct tagStore *store; char buf[BUFSIZ]; void *ptr; + ait_val_t *p = NULL; ioTRACE(2); @@ -208,41 +306,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, @@ -250,8 +341,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 */ @@ -259,13 +352,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 +372,7 @@ cmdUNSUBSCRIBE(void *srv, int len, void *arg) u_short mid = 0; register int i; struct tagStore *store, *tmp; + ait_val_t *p = NULL; ioTRACE(2); @@ -319,13 +412,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; @@ -336,7 +428,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); @@ -361,7 +453,6 @@ int cmdCONNECT(void *srv, int len, void *arg) { struct tagStore *store; - struct tagPkt *p; struct tagSession *sess = (struct tagSession*) arg; ioTRACE(2); @@ -392,12 +483,8 @@ cmdCONNECT(void *srv, int len, void *arg) 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.flag) + srv_Will(sess); if (sess->sess_will.msg) free(sess->sess_will.msg);