--- mqtt/src/mqttd_calls.c 2012/06/20 15:01:30 1.2.2.25 +++ mqtt/src/mqttd_calls.c 2012/06/26 08:05:58 1.2.2.26 @@ -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,7 +58,7 @@ 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; regex_t re; @@ -87,7 +76,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,13 +88,51 @@ 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 = NULL; + regex_t re; + regmatch_t match; + int ret, flg = 0; + 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.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.retain); + + 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)) { + flg = 1; /* MATCH */ + schedWrite(root, sendPacket, p, s->sess_sock, NULL, 0); + } + + regfree(&re); + } + } + + /* if subscriber not found then free packet */ + if (!flg) + freePkt(&p); + + call.DeletePUB_topic(&cfg, pub, sess->sess_cid, mid, psTopic, sess->sess_user, sess->sess_addr, -1); 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) { return 0; } @@ -120,7 +146,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 +160,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 +173,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 +200,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); @@ -221,7 +240,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); @@ -306,7 +325,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); @@ -362,7 +381,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); @@ -387,7 +406,6 @@ int cmdCONNECT(void *srv, int len, void *arg) { struct tagStore *store; - struct tagPkt *p; struct tagSession *sess = (struct tagSession*) arg; ioTRACE(2); @@ -416,13 +434,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)