version 1.2.2.22, 2012/06/19 08:32:17
|
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 struct tagPkt * | static inline ait_val_t * |
mkPkt(void * __restrict data, int dlen) |
mkPkt(void * __restrict data, int dlen) |
{ |
{ |
struct tagPkt *p = NULL; | ait_val_t *p = NULL; |
|
|
p = io_malloc(sizeof(struct tagPkt)); | if (!(p = io_allocVar())) { |
if (!p) { | |
ioDEBUG(7, "Error:: in send packet prepare #%d - %s", io_GetErrno(), io_GetError()); |
ioDEBUG(7, "Error:: in send packet prepare #%d - %s", io_GetErrno(), io_GetError()); |
return NULL; |
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) |
if (data && dlen > 0) |
AIT_SET_BUF(p->pkt_data, data, dlen); | AIT_SET_BUF(p, data, dlen); |
|
|
return p; |
return p; |
} |
} |
|
|
static inline void |
static inline void |
freePkt(struct tagPkt * __restrict p) | freePkt(ait_val_t ** __restrict p) |
{ |
{ |
if (!p) |
if (!p) |
return; |
return; |
|
|
io_freeVar(&p->pkt_data); | io_freeVar(p); |
io_free(p); | |
} |
} |
|
|
static void * |
static void * |
sendPacket(sched_task_t *task) |
sendPacket(sched_task_t *task) |
{ |
{ |
struct tagPkt *p = TASK_ARG(task); | ait_val_t *p = TASK_ARG(task); |
register int n, slen; |
register int n, slen; |
u_char *pos; |
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 ..."); |
ioDEBUG(9, "Error:: invalid packet or found empty content ..."); |
return NULL; |
return NULL; |
} |
} |
|
|
for (slen = AIT_LEN(p->pkt_data), pos = AIT_GET_BUF(p->pkt_data); slen > 0; | ioDEBUG(7, "Send packet length %d for socket %d\n", AIT_LEN(p), (u_int) TASK_FD(task)); |
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); |
n = send(TASK_FD(task), pos, slen, MSG_NOSIGNAL); |
if (n == -1) { |
if (n == -1) { |
ioSYSERR(0); |
ioSYSERR(0); |
Line 60 sendPacket(sched_task_t *task)
|
Line 97 sendPacket(sched_task_t *task)
|
} |
} |
} |
} |
|
|
freePkt(p); | freePkt(&p); |
return NULL; |
return NULL; |
} |
} |
|
|
/* --------------------------------------------------- */ |
|
|
|
static int |
static int |
pubOnce(struct tagSession *sess, char * __restrict psTopic, int datlen) | search4send(struct tagSession * __restrict sess, const char *topic, int datlen, char qos) |
{ |
{ |
struct tagPkt *p = NULL; |
|
struct tagSession *s = NULL; |
|
struct tagStore *st = NULL; |
|
regex_t re; |
regex_t re; |
regmatch_t match; |
regmatch_t match; |
int ret; | ait_val_t *p = NULL; |
| struct tagSession *s = NULL; |
| struct tagStore *st_, *st = NULL; |
char szStr[STRSIZ]; |
char szStr[STRSIZ]; |
|
int ret; |
|
|
|
assert(sess); |
|
|
TAILQ_FOREACH(s, &Sessions, sess_node) { |
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))) { | /* check for QoS */ |
regerror(ret, &re, szStr, sizeof szStr); | 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); |
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, sess->sess_sock, NULL, 0); |
|
} |
|
|
|
regfree(&re); |
|
} |
} |
} |
} |
|
|
return 0; |
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 |
|
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 |
static int |
pubAck(struct tagSession *sess, char * __restrict psTopic, int datlen) | pubOnce(struct tagSession *sess, char * __restrict psTopic, int datlen) |
{ |
{ |
return 0; | return search4send(sess, psTopic, datlen, MQTT_QOS_ONCE); |
} |
} |
|
|
static int |
static int |
pubExactly(struct tagSession *sess, char * __restrict psTopic, int datlen) | 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; |
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 |
int |
cmdPUBLISH(void *srv, int len, void *arg) |
cmdPUBLISH(void *srv, int len, void *arg) |
{ |
{ |
Line 119 cmdPUBLISH(void *srv, int len, void *arg)
|
Line 243 cmdPUBLISH(void *srv, int len, void *arg)
|
char szTopic[STRSIZ] = { 0 }; |
char szTopic[STRSIZ] = { 0 }; |
int siz = 0; |
int siz = 0; |
u_short mid = 0; |
u_short mid = 0; |
struct tagPkt *p = NULL; | ait_val_t *p = NULL; |
|
|
ioTRACE(2); |
ioTRACE(2); |
|
|
Line 133 cmdPUBLISH(void *srv, int len, void *arg)
|
Line 257 cmdPUBLISH(void *srv, int len, void *arg)
|
return 0; |
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; |
hdr = (struct mqtthdr*) sess->sess_buf->msg_base; |
switch (hdr->mqtt_msg.qos) { |
switch (hdr->mqtt_msg.qos) { |
case MQTT_QOS_ACK: |
case MQTT_QOS_ACK: |
pubAck(sess, szTopic, siz); | if (pubAck(sess, mid, szTopic, mqtt_pktLen(hdr))) |
| return 0; |
siz = mqtt_msgPUBACK(sess->sess_buf, mid); |
siz = mqtt_msgPUBACK(sess->sess_buf, mid); |
if (siz == -1) { |
if (siz == -1) { |
ioDEBUG(5, "Error:: in msgPUBACK #%d - %s", |
ioDEBUG(5, "Error:: in msgPUBACK #%d - %s", |
Line 154 cmdPUBLISH(void *srv, int len, void *arg)
|
Line 270 cmdPUBLISH(void *srv, int len, void *arg)
|
} |
} |
break; |
break; |
case MQTT_QOS_EXACTLY: |
case MQTT_QOS_EXACTLY: |
pubExactly(sess, szTopic, siz); | if (pubExactly(sess, mid, szTopic, mqtt_pktLen(hdr))) |
| return 0; |
siz = mqtt_msgPUBREC(sess->sess_buf, mid); |
siz = mqtt_msgPUBREC(sess->sess_buf, mid); |
if (siz == -1) { |
if (siz == -1) { |
ioDEBUG(5, "Error:: in msgPUBREC #%d - %s", |
ioDEBUG(5, "Error:: in msgPUBREC #%d - %s", |
Line 163 cmdPUBLISH(void *srv, int len, void *arg)
|
Line 280 cmdPUBLISH(void *srv, int len, void *arg)
|
} |
} |
break; |
break; |
case MQTT_QOS_ONCE: |
case MQTT_QOS_ONCE: |
pubOnce(sess, szTopic, siz); | pubOnce(sess, szTopic, mqtt_pktLen(hdr)); |
default: |
default: |
return 0; |
return 0; |
} |
} |
Line 180 cmdPUBREL(void *srv, int len, void *arg)
|
Line 297 cmdPUBREL(void *srv, int len, void *arg)
|
struct tagSession *sess = (struct tagSession*) arg; |
struct tagSession *sess = (struct tagSession*) arg; |
int siz = 0; |
int siz = 0; |
u_short mid = 0; |
u_short mid = 0; |
struct tagPkt *p = NULL; | ait_val_t *p = NULL; |
|
|
ioTRACE(2); |
ioTRACE(2); |
|
|
Line 194 cmdPUBREL(void *srv, int len, void *arg)
|
Line 311 cmdPUBREL(void *srv, int len, void *arg)
|
return 0; |
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); |
siz = mqtt_msgPUBCOMP(sess->sess_buf, mid); |
if (siz == -1) { |
if (siz == -1) { |
ioDEBUG(5, "Error:: in msgPUBCOMP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
ioDEBUG(5, "Error:: in msgPUBCOMP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
return 0; |
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); |
schedWrite(root, sendPacket, p, sess->sess_sock, NULL, 0); |
return 0; |
return 0; |
} |
} |
Line 220 cmdSUBSCRIBE(void *srv, int len, void *arg)
|
Line 338 cmdSUBSCRIBE(void *srv, int len, void *arg)
|
struct tagStore *store; |
struct tagStore *store; |
char buf[BUFSIZ]; |
char buf[BUFSIZ]; |
void *ptr; |
void *ptr; |
struct tagPkt *p = NULL; | ait_val_t *p = NULL; |
|
|
ioTRACE(2); |
ioTRACE(2); |
|
|
Line 236 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++) { |
/* convert topic to sql search statement */ | store = io_malloc(sizeof(struct tagStore)); |
if (mqtt_sqlTopic(subs[i].sub_topic.msg_base, buf, sizeof buf) == -1) { | if (!store) { |
ioDEBUG(5, "Error:: in db #%d - %s", mqtt_GetErrno(), mqtt_GetError()); | ioSYSERR(0); |
goto end; | continue; |
| } else { |
| store->st_msgid = mid; |
| mqtt_subCopy(&store->st_subscr, &subs[i]); |
| subs[i].sub_ret = MQTT_QOS_DENY; |
} |
} |
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 */ | /* add to cache */ |
SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node); | SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node); |
|
|
/* convert topic to regexp */ | /* 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()); | ioDEBUG(5, "Error:: in regexp #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
goto end; | } else { |
| ptr = realloc(store->st_subscr.sub_topic.msg_base, strlen(buf) + 1); |
| if (!ptr) { |
| ioSYSERR(0); |
| continue; |
} else { |
} else { |
ptr = realloc(store->st_subscr.sub_topic.msg_base, strlen(buf) + 1); | store->st_subscr.sub_topic.msg_base = ptr; |
if (!ptr) { | store->st_subscr.sub_topic.msg_len = strlen(buf) + 1; |
ioSYSERR(0); | memcpy(store->st_subscr.sub_topic.msg_base, buf, |
goto end; | store->st_subscr.sub_topic.msg_len); |
} 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 to db */ |
store->st_subscr.sub_topic.msg_base, | call.WritePUB_subscribe(&cfg, pub, sess->sess_cid, mid, buf, |
store->st_subscr.sub_topic.msg_len, sess->sess_addr); | 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 306 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; |
struct tagPkt *p = NULL; | ait_val_t *p = NULL; |
|
|
ioTRACE(2); |
ioTRACE(2); |
|
|
Line 362 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; |
struct tagPkt *p = NULL; | ait_val_t *p = NULL; |
|
|
ioTRACE(2); |
ioTRACE(2); |
|
|
Line 387 int
|
Line 501 int
|
cmdCONNECT(void *srv, int len, void *arg) |
cmdCONNECT(void *srv, int len, void *arg) |
{ |
{ |
struct tagStore *store; |
struct tagStore *store; |
struct tagPkt *p; |
|
struct tagSession *sess = (struct tagSession*) arg; |
struct tagSession *sess = (struct tagSession*) arg; |
|
|
ioTRACE(2); |
ioTRACE(2); |
Line 398 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); |
|
|
|
schedCancelby(root, taskTIMER, CRITERIA_CALL, sendRetain, NULL); |
|
|
if (sess->sess_clean) { |
if (sess->sess_clean) { |
if (call.FiniSessPUB) |
if (call.FiniSessPUB) |
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); |
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); |
Line 418 cmdCONNECT(void *srv, int len, void *arg)
|
Line 533 cmdCONNECT(void *srv, int len, void *arg)
|
io_free(store); |
io_free(store); |
} |
} |
|
|
while ((p = SLIST_FIRST(&sess->sess_sndpkt))) { | if (sess->sess_will.flag) |
SLIST_REMOVE_HEAD(&sess->sess_sndpkt, pkt_node); | pubWill(sess); |
| |
io_freeVar(&p->pkt_data); | |
io_free(p); | |
} | |
|
|
if (sess->sess_will.msg) |
if (sess->sess_will.msg) |
free(sess->sess_will.msg); |
free(sess->sess_will.msg); |