--- mqtt/src/mqttd_calls.c 2012/06/01 10:39:48 1.2.2.19 +++ mqtt/src/mqttd_calls.c 2017/10/08 22:49:25 1.5 @@ -1,149 +1,294 @@ +/************************************************************************* +* (C) 2011 AITNET ltd - Sofia/Bulgaria - +* by Michael Pounov +* +* $Author: misho $ +* $Id: mqttd_calls.c,v 1.5 2017/10/08 22:49:25 misho Exp $ +* +************************************************************************** +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 + +Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013 + by Michael Pounov . 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 +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 "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) { - ioDEBUG(7, "Error:: in send packet prepare #%d - %s", io_GetErrno(), io_GetError()); + if (!(p = ait_allocVar())) { + EVERBOSE(7, "Error:: in send packet prepare #%d - %s", + elwix_GetErrno(), elwix_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 || !*p) + if (!p) return; - io_freeVar(&(*p)->pkt_data); - io_free(*p); - *p = NULL; + ait_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)) { - ioDEBUG(9, "Error:: invalid packet or found empty content ..."); + if (!p || AIT_ISEMPTY(p)) { + EVERBOSE(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) { + EVERBOSE(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); + ESYSERR(0); break; } } + freePkt(&p); return NULL; } - static int -pubOnce(struct tagSession *sess, u_short mid, char * __restrict psTopic, - int topicLen, char * __restrict data, int datlen) +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); + EVERBOSE(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; + + ETRACE(); + + assert(task); + + sess = TASK_ARG(task); + assert(sess); + + if (!sess->sess_buf) { + EVERBOSE(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)); + EVERBOSE(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; + + ETRACE(); + + /* 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 -pubAck(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) { - return 0; + return search4send(sess, psTopic, datlen, MQTT_QOS_ONCE); } static int -pubExactly(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) { + 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) { 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); + ETRACE(); if (!sess) return -1; - ioDEBUG(5, "Exec PUBLISH session"); - siz = mqtt_readPUBLISH(sess->sess_buf, szTopic, sizeof szTopic, &mid, &data); + EVERBOSE(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()); + EVERBOSE(5, "Error:: in readPUBLISH #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } 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", + EVERBOSE(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", + EVERBOSE(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,33 +298,33 @@ 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); + ETRACE(); if (!sess) return -1; - ioDEBUG(5, "Exec PUBREL session"); + EVERBOSE(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()); + EVERBOSE(5, "Error:: in readPUBREL #%d - %s", mqtt_GetErrno(), mqtt_GetError()); 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()); + EVERBOSE(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,78 +339,74 @@ cmdSUBSCRIBE(void *srv, int len, void *arg) struct tagStore *store; char buf[BUFSIZ]; void *ptr; + ait_val_t *p = NULL; - ioTRACE(2); + ETRACE(); if (!sess) return -1; - ioDEBUG(5, "Exec SUBSCRIBE session"); + EVERBOSE(5, "Exec SUBSCRIBE session"); siz = mqtt_readSUBSCRIBE(sess->sess_buf, &mid, &subs); if (siz == -1) { - ioDEBUG(5, "Error:: in readSUBSCRIBE #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + EVERBOSE(5, "Error:: in readSUBSCRIBE #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } /* 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; + store = e_malloc(sizeof(struct tagStore)); + if (!store) { + ELIBERR(elwix); + 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 */ - 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) { + EVERBOSE(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) { + ESYSERR(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, - store->st_subscr.sub_topic.msg_base, - store->st_subscr.sub_topic.msg_len, sess->sess_addr); - + /* store to db */ + 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; - } 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 */ siz = mqtt_msgSUBACK(sess->sess_buf, subs, mid); if (siz == -1) { - ioDEBUG(5, "Error:: in msgSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + EVERBOSE(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,16 +421,17 @@ 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); + ETRACE(); if (!sess) return -1; - ioDEBUG(5, "Exec UNSUBSCRIBE session"); + EVERBOSE(5, "Exec UNSUBSCRIBE session"); siz = mqtt_readUNSUBSCRIBE(sess->sess_buf, &mid, &subs); if (siz == -1) { - ioDEBUG(5, "Error:: in readUNSUBSCRIBE #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + EVERBOSE(5, "Error:: in readUNSUBSCRIBE #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } @@ -306,7 +448,7 @@ cmdUNSUBSCRIBE(void *srv, int len, void *arg) 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); + e_free(store); } } @@ -317,15 +459,14 @@ cmdUNSUBSCRIBE(void *srv, int len, void *arg) /* send acknowledge */ siz = mqtt_msgUNSUBACK(sess->sess_buf, mid); if (siz == -1) { - ioDEBUG(5, "Error:: in msgUNSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + EVERBOSE(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,17 +477,17 @@ 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); + ETRACE(); if (!sess) return -1; - ioDEBUG(5, "Exec PINGREQ session"); + EVERBOSE(5, "Exec PINGREQ session"); siz = mqtt_msgPINGRESP(sess->sess_buf); if (siz == -1) { - ioDEBUG(5, "Error:: in msgPINGRESP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + EVERBOSE(5, "Error:: in msgPINGRESP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } else { p = mkPkt(sess->sess_buf->msg_base, siz); @@ -361,17 +502,18 @@ int cmdCONNECT(void *srv, int len, void *arg) { struct tagStore *store; - struct tagPkt *p; struct tagSession *sess = (struct tagSession*) arg; - ioTRACE(2); + ETRACE(); if (!sess) return -1; - ioDEBUG(5, "Exec CONNECT session"); + EVERBOSE(5, "Exec CONNECT session"); TAILQ_REMOVE(&Sessions, sess, sess_node); + schedCancelby(root, taskTIMER, CRITERIA_CALL, sendRetain, NULL); + if (sess->sess_clean) { if (call.FiniSessPUB) call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); @@ -389,16 +531,12 @@ cmdCONNECT(void *srv, int len, void *arg) if (store->st_subscr.sub_value.msg_base) free(store->st_subscr.sub_value.msg_base); - io_free(store); + e_free(store); } - while ((p = SLIST_FIRST(&sess->sess_sndpkt))) { - SLIST_REMOVE_HEAD(&sess->sess_sndpkt, pkt_node); + if (sess->sess_will.flag) + pubWill(sess); - io_freeVar(&p->pkt_data); - io_free(p); - } - if (sess->sess_will.msg) free(sess->sess_will.msg); if (sess->sess_will.topic) @@ -415,12 +553,12 @@ cmdDISCONNECT(void *srv, int len, void *arg) { struct tagSession *sess = (struct tagSession*) arg; - ioTRACE(2); + ETRACE(); if (!sess) return -1; - ioDEBUG(5, "Exec DISCONNECT session"); + EVERBOSE(5, "Exec DISCONNECT session"); call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, sess->sess_addr, sess->sess_user);