version 1.2.2.13, 2012/05/08 13:04:02
|
version 1.3.2.1, 2012/07/03 12:22:56
|
Line 1
|
Line 1
|
|
/************************************************************************* |
|
* (C) 2011 AITNET ltd - Sofia/Bulgaria - <misho@aitbg.com> |
|
* by Michael Pounov <misho@openbsd-bg.org> |
|
* |
|
* $Author$ |
|
* $Id$ |
|
* |
|
************************************************************************** |
|
The ELWIX and AITNET software is distributed under the following |
|
terms: |
|
|
|
All of the documentation and software included in the ELWIX and AITNET |
|
Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org> |
|
|
|
Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012 |
|
by Michael Pounov <misho@elwix.org>. All rights reserved. |
|
|
|
Redistribution and use in source and binary forms, with or without |
|
modification, are permitted provided that the following conditions |
|
are met: |
|
1. Redistributions of source code must retain the above copyright |
|
notice, this list of conditions and the following disclaimer. |
|
2. Redistributions in binary form must reproduce the above copyright |
|
notice, this list of conditions and the following disclaimer in the |
|
documentation and/or other materials provided with the distribution. |
|
3. All advertising materials mentioning features or use of this software |
|
must display the following acknowledgement: |
|
This product includes software developed by Michael Pounov <misho@elwix.org> |
|
ELWIX - Embedded LightWeight unIX and its contributors. |
|
4. Neither the name of AITNET nor the names of its contributors |
|
may be used to endorse or promote products derived from this software |
|
without specific prior written permission. |
|
|
|
THIS SOFTWARE IS PROVIDED BY AITNET AND CONTRIBUTORS ``AS IS'' AND |
|
ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE |
|
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE |
|
ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE |
|
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL |
|
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS |
|
OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) |
|
HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT |
|
LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY |
|
OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF |
|
SUCH DAMAGE. |
|
*/ |
#include "global.h" |
#include "global.h" |
#include "mqttd.h" |
#include "mqttd.h" |
|
#include "utils.h" |
#include "rtlm.h" |
#include "rtlm.h" |
#include "mqttd_calls.h" |
#include "mqttd_calls.h" |
|
|
|
|
|
static inline ait_val_t * |
|
mkPkt(void * __restrict data, int dlen) |
|
{ |
|
ait_val_t *p = NULL; |
|
|
|
if (!(p = io_allocVar())) { |
|
ioDEBUG(7, "Error:: in send packet prepare #%d - %s", io_GetErrno(), io_GetError()); |
|
return NULL; |
|
} |
|
|
|
if (data && dlen > 0) |
|
AIT_SET_BUF(p, data, dlen); |
|
|
|
return p; |
|
} |
|
|
|
static inline void |
|
freePkt(ait_val_t ** __restrict p) |
|
{ |
|
if (!p) |
|
return; |
|
|
|
io_freeVar(p); |
|
} |
|
|
|
static void * |
|
sendPacket(sched_task_t *task) |
|
{ |
|
ait_val_t *p = TASK_ARG(task); |
|
register int n, slen; |
|
u_char *pos; |
|
|
|
if (!p || AIT_ISEMPTY(p)) { |
|
ioDEBUG(9, "Error:: invalid packet or found empty content ..."); |
|
return NULL; |
|
} |
|
|
|
ioDEBUG(7, "Send packet length %d for socket %d\n", AIT_LEN(p), (u_int) TASK_FD(task)); |
|
|
|
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); |
|
break; |
|
} |
|
} |
|
|
|
freePkt(&p); |
|
return NULL; |
|
} |
|
|
|
static int |
|
search4send(struct tagSession * __restrict sess, const char *topic, int datlen, char qos) |
|
{ |
|
regex_t re; |
|
regmatch_t match; |
|
ait_val_t *p = NULL; |
|
struct tagSession *s = NULL; |
|
struct tagStore *st_, *st = NULL; |
|
char szStr[STRSIZ]; |
|
int ret; |
|
|
|
assert(sess); |
|
|
|
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 >= qos) { |
|
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, topic, 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; |
|
} |
|
|
|
/* --------------------------------------------------- */ |
|
|
|
void * |
|
sendRetain(sched_task_t *task) |
|
{ |
|
mqtt_subscr_t *subs, *s; |
|
struct tagSession *sess; |
|
int siz; |
|
|
|
ioTRACE(2); |
|
|
|
assert(task); |
|
|
|
sess = TASK_ARG(task); |
|
assert(sess); |
|
|
|
if (!sess->sess_buf) { |
|
ioDEBUG(9, "WARNING! No allocated buffer!?!\n"); |
|
return NULL; |
|
} |
|
|
|
subs = call.ReadPUB_topic(&cfg, pub, "%", "%", 1); |
|
if (!subs) |
|
return NULL; |
|
|
|
for (s = subs; s && s->sub_topic.msg_base; s++) { |
|
siz = s->sub_value.msg_len; |
|
memcpy(sess->sess_buf->msg_base, s->sub_value.msg_base, |
|
MIN(sess->sess_buf->msg_len, s->sub_value.msg_len)); |
|
ioDEBUG(7, "Sending retain message %d bytes, QoS %hhd topic '%s' data length %d\n", |
|
siz, s->sub_ret, (char*) s->sub_topic.msg_base, s->sub_value.msg_len); |
|
if (siz > 0) |
|
search4send(sess, s->sub_topic.msg_base, siz, s->sub_ret); |
|
} |
|
|
|
mqtt_subFree(&subs); |
|
return NULL; |
|
} |
|
|
int |
int |
|
pubWill(struct tagSession * __restrict sess) |
|
{ |
|
int datlen; |
|
|
|
ioTRACE(2); |
|
|
|
/* prepare will packet */ |
|
datlen = mqtt_msgPUBLISH(sess->sess_buf, sess->sess_will.topic, 0xDEAD, 0, 1, 0, |
|
sess->sess_will.msg, sess->sess_will.msg ? strlen(sess->sess_will.msg) : 0); |
|
if (datlen == -1) |
|
return -1; /* error */ |
|
|
|
return search4send(sess, sess->sess_will.topic, datlen, MQTT_QOS_ACK); |
|
} |
|
|
|
static int |
|
pubOnce(struct tagSession *sess, char * __restrict psTopic, int datlen) |
|
{ |
|
return search4send(sess, psTopic, datlen, MQTT_QOS_ONCE); |
|
} |
|
|
|
static int |
|
pubAck(struct tagSession *sess, u_short mid, char * __restrict psTopic, int datlen) |
|
{ |
|
struct mqtthdr *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, |
|
sess->sess_buf->msg_base, datlen, sess->sess_user, sess->sess_addr, |
|
hdr->mqtt_msg.qos, hdr->mqtt_msg.retain); |
|
|
|
search4send(sess, psTopic, datlen, MQTT_QOS_ACK); |
|
|
|
/* delete not retain message */ |
|
call.DeletePUB_topic(&cfg, pub, sess->sess_cid, mid, psTopic, sess->sess_user, |
|
sess->sess_addr, 0); |
|
return 0; |
|
} |
|
|
|
static int |
|
pubExactly(struct tagSession *sess, u_short mid, char * __restrict psTopic, int datlen) |
|
{ |
|
struct mqtthdr *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, |
|
sess->sess_buf->msg_base, datlen, sess->sess_user, sess->sess_addr, |
|
hdr->mqtt_msg.qos, hdr->mqtt_msg.retain); |
|
|
|
return search4send(sess, psTopic, datlen, MQTT_QOS_EXACTLY); |
|
} |
|
|
|
|
|
int |
cmdPUBLISH(void *srv, int len, void *arg) |
cmdPUBLISH(void *srv, int len, void *arg) |
{ |
{ |
struct mqtthdr *hdr; |
struct mqtthdr *hdr; |
struct tagSession *sess = (struct tagSession*) arg; |
struct tagSession *sess = (struct tagSession*) arg; |
|
char szTopic[STRSIZ] = { 0 }; |
|
int siz = 0; |
|
u_short mid = 0; |
|
ait_val_t *p = NULL; |
|
|
ioTRACE(2); |
ioTRACE(2); |
|
|
if (!sess) |
if (!sess) |
return -1; |
return -1; |
|
|
|
ioDEBUG(5, "Exec PUBLISH session"); |
|
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; |
|
} |
|
|
hdr = (struct mqtthdr*) sess->sess_buf->msg_base; |
hdr = (struct mqtthdr*) sess->sess_buf->msg_base; |
switch (hdr->mqtt_msg.qos) { |
switch (hdr->mqtt_msg.qos) { |
case MQTT_QOS_ONCE: |
|
break; |
|
case MQTT_QOS_ACK: |
case MQTT_QOS_ACK: |
|
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()); |
|
return 0; |
|
} |
break; |
break; |
case MQTT_QOS_EXACTLY: |
case MQTT_QOS_EXACTLY: |
|
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()); |
|
return 0; |
|
} |
break; |
break; |
|
case MQTT_QOS_ONCE: |
|
pubOnce(sess, szTopic, mqtt_pktLen(hdr)); |
default: |
default: |
ioDEBUG(1, "Error:: Unknown QoS %d - rejected publishing request", |
|
hdr->mqtt_msg.qos); |
|
return 0; |
return 0; |
} |
} |
|
|
|
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; |
return 0; |
} |
} |
|
|
int |
int |
cmdPUBREL(void *srv, int len, void *arg) |
cmdPUBREL(void *srv, int len, void *arg) |
{ |
{ |
struct mqtthdr *hdr; |
|
struct tagSession *sess = (struct tagSession*) arg; |
struct tagSession *sess = (struct tagSession*) arg; |
|
int siz = 0; |
|
u_short mid = 0; |
|
ait_val_t *p = NULL; |
|
|
ioTRACE(2); |
ioTRACE(2); |
|
|
if (!sess) |
if (!sess) |
return -1; |
return -1; |
|
|
hdr = (struct mqtthdr*) sess->sess_buf->msg_base; | ioDEBUG(5, "Exec PUBREL session"); |
| mid = mqtt_readPUBREL(sess->sess_buf); |
| if (mid == (u_short) -1) { |
| ioDEBUG(5, "Error:: in readPUBREL #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
| return 0; |
| } |
|
|
|
/* 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; |
|
} |
|
|
|
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; |
return 0; |
} |
} |
|
|
Line 57 cmdSUBSCRIBE(void *srv, int len, void *arg)
|
Line 336 cmdSUBSCRIBE(void *srv, int len, void *arg)
|
u_short mid = 0; |
u_short mid = 0; |
register int i; |
register int i; |
struct tagStore *store; |
struct tagStore *store; |
|
char buf[BUFSIZ]; |
|
void *ptr; |
|
ait_val_t *p = NULL; |
|
|
ioTRACE(2); |
ioTRACE(2); |
|
|
Line 72 cmdSUBSCRIBE(void *srv, int len, void *arg)
|
Line 354 cmdSUBSCRIBE(void *srv, int len, void *arg)
|
|
|
/* add to db */ |
/* add to db */ |
for (i = 0; i < siz; i++) { |
for (i = 0; i < siz; i++) { |
if (call.WritePUB_subscribe(&cfg, pub, mid, subs[i].sub_topic.msg_base, | store = io_malloc(sizeof(struct tagStore)); |
sess->sess_user, sess->sess_addr, subs[i].sub_ret) > 0) { | if (!store) { |
store = malloc(sizeof(struct tagStore)); | ioSYSERR(0); |
if (!store) { | continue; |
| } else { |
| store->st_msgid = mid; |
| mqtt_subCopy(&store->st_subscr, &subs[i]); |
| subs[i].sub_ret = MQTT_QOS_DENY; |
| } |
| |
| /* 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, 1) == -1) { |
| ioDEBUG(5, "Error:: in regexp #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
| } else { |
| ptr = realloc(store->st_subscr.sub_topic.msg_base, strlen(buf) + 1); |
| if (!ptr) { |
ioSYSERR(0); |
ioSYSERR(0); |
goto end; | continue; |
} else { |
} else { |
store->st_msgid = mid; | store->st_subscr.sub_topic.msg_base = ptr; |
mqtt_subCopy(&store->st_subscr, &subs[i]); | 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); |
} |
} |
|
|
/* add to cache */ | /* store to db */ |
SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node); | call.WritePUB_subscribe(&cfg, pub, sess->sess_cid, mid, buf, |
| sess->sess_user, sess->sess_addr, store->st_subscr.sub_ret); |
| /* subscribe pass */ |
subs[i].sub_ret = MQTT_QOS_PASS; |
subs[i].sub_ret = MQTT_QOS_PASS; |
} else | |
subs[i].sub_ret = MQTT_QOS_DENY; | call.LOG(logg, "Added [%s] SUBSCRIBE '%s'(%d) QoS=%d from %s\n", sess->sess_cid, |
| store->st_subscr.sub_topic.msg_base, |
| store->st_subscr.sub_topic.msg_len, |
| store->st_subscr.sub_ret, sess->sess_addr); |
| } |
} |
} |
|
|
/* send acknowledge */ |
/* send acknowledge */ |
Line 96 cmdSUBSCRIBE(void *srv, int len, void *arg)
|
Line 400 cmdSUBSCRIBE(void *srv, int len, void *arg)
|
if (siz == -1) { |
if (siz == -1) { |
ioDEBUG(5, "Error:: in msgSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
ioDEBUG(5, "Error:: in msgSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
goto end; |
goto end; |
} | } else { |
if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, MSG_NOSIGNAL)) == -1) | p = mkPkt(sess->sess_buf->msg_base, siz); |
ioSYSERR(0); | |
else { | |
ioDEBUG(5, "Sended %d bytes.", siz); | |
memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); |
memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); |
} |
} |
|
|
|
schedWrite(root, sendPacket, p, sess->sess_sock, NULL, 0); |
end: |
end: |
mqtt_subFree(&subs); |
mqtt_subFree(&subs); |
return 0; |
return 0; |
Line 117 cmdUNSUBSCRIBE(void *srv, int len, void *arg)
|
Line 420 cmdUNSUBSCRIBE(void *srv, int len, void *arg)
|
u_short mid = 0; |
u_short mid = 0; |
register int i; |
register int i; |
struct tagStore *store, *tmp; |
struct tagStore *store, *tmp; |
|
ait_val_t *p = NULL; |
|
|
ioTRACE(2); |
ioTRACE(2); |
|
|
Line 138 cmdUNSUBSCRIBE(void *srv, int len, void *arg)
|
Line 442 cmdUNSUBSCRIBE(void *srv, int len, void *arg)
|
!strcmp(store->st_subscr.sub_topic.msg_base, |
!strcmp(store->st_subscr.sub_topic.msg_base, |
subs[i].sub_topic.msg_base)) { |
subs[i].sub_topic.msg_base)) { |
SLIST_REMOVE(&sess->sess_subscr, store, tagStore, st_node); |
SLIST_REMOVE(&sess->sess_subscr, store, tagStore, st_node); |
free(store); | |
| if (store->st_subscr.sub_topic.msg_base) |
| free(store->st_subscr.sub_topic.msg_base); |
| if (store->st_subscr.sub_value.msg_base) |
| free(store->st_subscr.sub_value.msg_base); |
| io_free(store); |
} |
} |
} |
} |
|
|
call.DeletePUB_subscribe(&cfg, pub, subs[i].sub_topic.msg_base, | call.DeletePUB_subscribe(&cfg, pub, sess->sess_cid, subs[i].sub_topic.msg_base, |
sess->sess_user, sess->sess_addr); | sess->sess_user, "%"); |
} |
} |
|
|
/* send acknowledge */ |
/* send acknowledge */ |
Line 151 cmdUNSUBSCRIBE(void *srv, int len, void *arg)
|
Line 460 cmdUNSUBSCRIBE(void *srv, int len, void *arg)
|
if (siz == -1) { |
if (siz == -1) { |
ioDEBUG(5, "Error:: in msgUNSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
ioDEBUG(5, "Error:: in msgUNSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
goto end; |
goto end; |
} | } else { |
if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, MSG_NOSIGNAL)) == -1) | p = mkPkt(sess->sess_buf->msg_base, siz); |
ioSYSERR(0); | |
else { | |
ioDEBUG(5, "Sended %d bytes.", siz); | |
memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); |
memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); |
} |
} |
|
|
|
schedWrite(root, sendPacket, p, sess->sess_sock, NULL, 0); |
end: |
end: |
mqtt_subFree(&subs); |
mqtt_subFree(&subs); |
return 0; |
return 0; |
Line 168 cmdPINGREQ(void *srv, int len, void *arg)
|
Line 476 cmdPINGREQ(void *srv, int len, void *arg)
|
{ |
{ |
struct tagSession *sess = (struct tagSession*) arg; |
struct tagSession *sess = (struct tagSession*) arg; |
int siz = 0; |
int siz = 0; |
|
ait_val_t *p = NULL; |
|
|
ioTRACE(2); |
ioTRACE(2); |
|
|
Line 179 cmdPINGREQ(void *srv, int len, void *arg)
|
Line 488 cmdPINGREQ(void *srv, int len, void *arg)
|
if (siz == -1) { |
if (siz == -1) { |
ioDEBUG(5, "Error:: in msgPINGRESP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
ioDEBUG(5, "Error:: in msgPINGRESP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
return 0; |
return 0; |
} |
|
if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, MSG_NOSIGNAL)) == -1) { |
|
ioSYSERR(0); |
|
return 0; |
|
} else { |
} else { |
ioDEBUG(5, "Sended %d bytes.", siz); | p = mkPkt(sess->sess_buf->msg_base, siz); |
memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); |
memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); |
} |
} |
|
|
|
schedWrite(root, sendPacket, p, sess->sess_sock, NULL, 0); |
return 0; |
return 0; |
} |
} |
|
|
Line 205 cmdCONNECT(void *srv, int len, void *arg)
|
Line 511 cmdCONNECT(void *srv, int len, void *arg)
|
ioDEBUG(5, "Exec CONNECT session"); |
ioDEBUG(5, "Exec CONNECT session"); |
TAILQ_REMOVE(&Sessions, sess, sess_node); |
TAILQ_REMOVE(&Sessions, sess, sess_node); |
|
|
if (call.FiniSessPUB) | schedCancelby(root, taskTIMER, CRITERIA_CALL, sendRetain, NULL); |
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); | |
|
|
|
if (sess->sess_clean) { |
|
if (call.FiniSessPUB) |
|
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); |
|
if (call.DeletePUB_subscribe) |
|
call.DeletePUB_subscribe(&cfg, pub, sess->sess_cid, "%", sess->sess_user, "%"); |
|
if (call.WipePUB_topic) |
|
call.WipePUB_topic(&cfg, pub, sess->sess_cid, sess->sess_user, -1); |
|
} |
|
|
while ((store = SLIST_FIRST(&sess->sess_subscr))) { |
while ((store = SLIST_FIRST(&sess->sess_subscr))) { |
SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node); |
SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node); |
|
|
Line 216 cmdCONNECT(void *srv, int len, void *arg)
|
Line 530 cmdCONNECT(void *srv, int len, void *arg)
|
if (store->st_subscr.sub_value.msg_base) |
if (store->st_subscr.sub_value.msg_base) |
free(store->st_subscr.sub_value.msg_base); |
free(store->st_subscr.sub_value.msg_base); |
|
|
free(store); | io_free(store); |
} |
} |
|
|
|
if (sess->sess_will.flag) |
|
pubWill(sess); |
|
|
if (sess->sess_will.msg) |
if (sess->sess_will.msg) |
free(sess->sess_will.msg); |
free(sess->sess_will.msg); |