| version 1.2.2.25, 2012/05/05 14:51:02 | version 1.5.2.1, 2013/01/18 10:17:22 | 
| 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, 2013 | 
 |  | 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 "rtlm.h" | #include "rtlm.h" | 
 | #include "utils.h" |  | 
 | #include "mqttd.h" | #include "mqttd.h" | 
 | #include "mqttd_calls.h" | #include "mqttd_calls.h" | 
 |  | #include "utils.h" | 
 |  |  | 
 |  |  | 
 | static inline struct tagSession * | static inline struct tagSession * | 
| Line 11  initSession(int sock, ait_val_t * __restrict v) | Line 56  initSession(int sock, ait_val_t * __restrict v) | 
 | struct tagSession *sess = NULL; | struct tagSession *sess = NULL; | 
 | const char *str; | const char *str; | 
 |  |  | 
| ioTRACE(5); | ETRACE(5); | 
 |  |  | 
 | if (!v) | if (!v) | 
 | return NULL; | return NULL; | 
 |  |  | 
| sess = malloc(sizeof(struct tagSession)); | sess = e_malloc(sizeof(struct tagSession)); | 
 | if (!sess) { | if (!sess) { | 
| ioSYSERR(0); | ELIBERR(elwix); | 
 | return NULL; | return NULL; | 
 | } else | } else | 
 | memset(sess, 0, sizeof(struct tagSession)); | memset(sess, 0, sizeof(struct tagSession)); | 
| Line 33  initSession(int sock, ait_val_t * __restrict v) | Line 78  initSession(int sock, ait_val_t * __restrict v) | 
 |  |  | 
 | sess->sess_buf = mqtt_msgAlloc(USHRT_MAX); | sess->sess_buf = mqtt_msgAlloc(USHRT_MAX); | 
 | if (!sess->sess_buf) { | if (!sess->sess_buf) { | 
| ioLIBERR(mqtt); | ELIBERR(mqtt); | 
| free(sess); | e_free(sess); | 
 | return NULL; | return NULL; | 
 | } | } | 
 |  |  | 
 | /* init server actor */ | /* init server actor */ | 
| sess->sess_srv = mqtt_srv_Init(sock, sess->sess_buf); | sess->sess_srv = mqtt_srv_cliInit(sock, sess->sess_buf, sess->sess_ka, 1); | 
 | if (!sess->sess_srv) { | if (!sess->sess_srv) { | 
| ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError()); | EVERBOSE(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError()); | 
 | mqtt_msgFree(&sess->sess_buf, 42); | mqtt_msgFree(&sess->sess_buf, 42); | 
| free(sess); | e_free(sess); | 
 | return NULL; | return NULL; | 
 | } else { | } else { | 
 | mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_CONNECT, cmdCONNECT); | mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_CONNECT, cmdCONNECT); | 
| Line 65  finiSession(struct tagSession *sess) | Line 110  finiSession(struct tagSession *sess) | 
 | { | { | 
 | struct tagStore *store; | struct tagStore *store; | 
 |  |  | 
| ioTRACE(5); | ETRACE(5); | 
 |  |  | 
 | if (!sess) | if (!sess) | 
 | return; | return; | 
 |  |  | 
| 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) /* only dynamic messages */ | 
 |  | call.WipePUB_topic(&cfg, pub, sess->sess_cid, sess->sess_user, 0); | 
 |  | } | 
 |  |  | 
 | 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 81  finiSession(struct tagSession *sess) | Line 134  finiSession(struct tagSession *sess) | 
 | 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); | e_free(store); | 
 | } | } | 
 |  |  | 
| if (sess->sess_will.msg) | if (sess->sess_will.flag) | 
| free(sess->sess_will.msg); | pubWill(sess); | 
|  |  | 
 | if (sess->sess_will.topic) | if (sess->sess_will.topic) | 
 | free(sess->sess_will.topic); | free(sess->sess_will.topic); | 
 |  | if (sess->sess_will.msg) | 
 |  | free(sess->sess_will.msg); | 
 |  |  | 
 | if (sess->sess_sock > STDERR_FILENO) | if (sess->sess_sock > STDERR_FILENO) | 
 | srv_Close(sess->sess_sock); | srv_Close(sess->sess_sock); | 
 |  |  | 
| mqtt_srv_Fini(&sess->sess_srv); | mqtt_srv_cliFini(&sess->sess_srv); | 
 | mqtt_msgFree(&sess->sess_buf, 42); | mqtt_msgFree(&sess->sess_buf, 42); | 
 |  |  | 
| free(sess); | e_free(sess); | 
 | } | } | 
 |  |  | 
 | static void * | static void * | 
| Line 104  leaveClient(sched_task_t *task) | Line 160  leaveClient(sched_task_t *task) | 
 | struct tagSession *sess; | struct tagSession *sess; | 
 | int ret; | int ret; | 
 |  |  | 
| ioTRACE(4); | ETRACE(4); | 
 |  |  | 
 | assert(task); | assert(task); | 
 |  |  | 
| Line 116  leaveClient(sched_task_t *task) | Line 172  leaveClient(sched_task_t *task) | 
 | ret = mqtt_msgDISCONNECT(sess->sess_buf); | ret = mqtt_msgDISCONNECT(sess->sess_buf); | 
 | send(TASK_FD(task), sess->sess_buf->msg_base, ret, MSG_NOSIGNAL); | send(TASK_FD(task), sess->sess_buf->msg_base, ret, MSG_NOSIGNAL); | 
 |  |  | 
| ioDEBUG(1, "Close socket=%d", TASK_FD(task)); | EVERBOSE(1, "Close socket=%ld", (long) TASK_FD(task)); | 
 | call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, | call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, | 
 | sess->sess_addr, sess->sess_user); | sess->sess_addr, sess->sess_user); | 
 |  |  | 
| Line 130  dispatchSession(sched_task_t *task) | Line 186  dispatchSession(sched_task_t *task) | 
 | int ret, len = 0; | int ret, len = 0; | 
 | struct tagSession *sess; | struct tagSession *sess; | 
 |  |  | 
| ioTRACE(2); | ETRACE(2); | 
 |  |  | 
 | assert(task); | assert(task); | 
 |  |  | 
| Line 139  dispatchSession(sched_task_t *task) | Line 195  dispatchSession(sched_task_t *task) | 
 |  |  | 
 | /* receive & decode packet */ | /* receive & decode packet */ | 
 | if ((ret = recv(TASK_FD(task), sess->sess_buf->msg_base, sess->sess_buf->msg_len, 0)) == -1) { | if ((ret = recv(TASK_FD(task), sess->sess_buf->msg_base, sess->sess_buf->msg_len, 0)) == -1) { | 
| ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); | EVERBOSE(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); | 
|  | TAILQ_REMOVE(&Sessions, sess, sess_node); | 
 | finiSession(sess); | finiSession(sess); | 
 | return NULL; | return NULL; | 
 | } else if (!ret) { | } else if (!ret) { | 
| ioDEBUG(4, "Session %s EOF received.", sess->sess_cid); | EVERBOSE(4, "Session %s EOF received.", sess->sess_cid); | 
|  | TAILQ_REMOVE(&Sessions, sess, sess_node); | 
 | finiSession(sess); | finiSession(sess); | 
 | return NULL; | return NULL; | 
 | } | } | 
 |  |  | 
 | do { | do { | 
 | /* dispatch message type */ | /* dispatch message type */ | 
| if ((len = mqtt_srv_Dispatch(sess->sess_srv, ret, sess)) < 0) { | if ((len = mqtt_srv_cliDispatch(sess->sess_srv, ret, sess)) < 0) { | 
| if (len == -1) | if (len == -1) { | 
| ioLIBERR(mqtt); | ELIBERR(mqtt); | 
| finiSession(sess); | finiSession(sess); | 
|  | } else if (len == -2) { | 
|  | TAILQ_REMOVE(&Sessions, sess, sess_node); | 
|  | finiSession(sess); | 
|  | } else if (len == -3) | 
|  | schedEvent(root, startSession, NULL, (u_long) TASK_FD(task), sess, ret); | 
 | } else | } else | 
 | ret -= len; | ret -= len; | 
 | } while (len > 0 && ret > 0); | } while (len > 0 && ret > 0); | 
 |  |  | 
 | if (len >= 0 && !schedRead(root, dispatchSession, TASK_ARG(task), TASK_FD(task), NULL, 0)) { | if (len >= 0 && !schedRead(root, dispatchSession, TASK_ARG(task), TASK_FD(task), NULL, 0)) { | 
| ioLIBERR(sched); | ELIBERR(sched); | 
|  | TAILQ_REMOVE(&Sessions, sess, sess_node); | 
 | finiSession(sess); | finiSession(sess); | 
 | } | } | 
 | return NULL; | return NULL; | 
 | } | } | 
 |  |  | 
 |  | static int | 
 |  | loadSubscribes(struct tagSession * __restrict sess, mqtt_subscr_t * __restrict subs) | 
 |  | { | 
 |  | register int i; | 
 |  | struct tagStore *store; | 
 |  |  | 
 |  | if (!subs) | 
 |  | return -1; | 
 |  |  | 
 |  | for (i = 0; subs[i].sub_topic.msg_base; i++) { | 
 |  | store = e_malloc(sizeof(struct tagStore)); | 
 |  | if (!store) { | 
 |  | ELIBERR(elwix); | 
 |  | continue; | 
 |  | } else { | 
 |  | store->st_msgid = 0; | 
 |  | mqtt_subCopy(&store->st_subscr, &subs[i]); | 
 |  | } | 
 |  |  | 
 |  | /* add to cache */ | 
 |  | SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node); | 
 |  | } | 
 |  |  | 
 |  | return 0; | 
 |  | } | 
 |  |  | 
 | void * | void * | 
 | startSession(sched_task_t *task) | startSession(sched_task_t *task) | 
 | { | { | 
| Line 172  startSession(sched_task_t *task) | Line 262  startSession(sched_task_t *task) | 
 | mqtt_msg_t buf = { basebuf, sizeof basebuf }; | mqtt_msg_t buf = { basebuf, sizeof basebuf }; | 
 | mqtthdr_connflgs_t flg; | mqtthdr_connflgs_t flg; | 
 | mqtthdr_connack_t cack; | mqtthdr_connack_t cack; | 
 |  | ait_val_t *v; | 
 | struct tagSession *s, *sess = NULL; | struct tagSession *s, *sess = NULL; | 
 | int ret, wlen; | int ret, wlen; | 
 |  | mqtt_subscr_t *subs; | 
 |  | struct timespec ts = { RETAIN_TIMEOUT, 0 }; | 
 |  | intptr_t sock; | 
 |  |  | 
| ioTRACE(4); | ETRACE(4); | 
 |  |  | 
 | assert(task); | assert(task); | 
 |  |  | 
 | if (!TASK_DATA(task)) { | if (!TASK_DATA(task)) { | 
 |  | v = TASK_ARG(task); | 
 | /* flow from accept new clients */ | /* flow from accept new clients */ | 
| sess = initSession(TASK_FD(task), TASK_ARG(task)); | sess = initSession(TASK_FD(task), v); | 
| io_freeVar(TASK_ARG(task)); | ait_freeVar(&v); | 
 | if (!sess) { | if (!sess) { | 
 | close(TASK_FD(task)); | close(TASK_FD(task)); | 
 | return NULL; | return NULL; | 
| Line 190  startSession(sched_task_t *task) | Line 285  startSession(sched_task_t *task) | 
 |  |  | 
 | /* receive & decode packet */ | /* receive & decode packet */ | 
 | if (recv(sess->sess_sock, buf.msg_base, buf.msg_len, 0) == -1) { | if (recv(sess->sess_sock, buf.msg_base, buf.msg_len, 0) == -1) { | 
| ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); | EVERBOSE(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); | 
 | finiSession(sess); | finiSession(sess); | 
 | return NULL; | return NULL; | 
 | } | } | 
| Line 208  startSession(sched_task_t *task) | Line 303  startSession(sched_task_t *task) | 
 | ret = cack.retcode; | ret = cack.retcode; | 
 | flg.flags = cack.reserved; | flg.flags = cack.reserved; | 
 | if (flg.reserved) { | if (flg.reserved) { | 
| ioDEBUG(3, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError()); | EVERBOSE(3, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError()); | 
 | goto end; | goto end; | 
 | } else { | } else { | 
 | sess->sess_clean = flg.clean_sess; | sess->sess_clean = flg.clean_sess; | 
 | sess->sess_will.qos = flg.will_qos; | sess->sess_will.qos = flg.will_qos; | 
 | sess->sess_will.retain = flg.will_retain; | sess->sess_will.retain = flg.will_retain; | 
 | sess->sess_will.flag = flg.will_flg; | sess->sess_will.flag = flg.will_flg; | 
 |  |  | 
 |  | sess->sess_srv->timeout = sess->sess_ka; | 
 | } | } | 
 |  |  | 
 | /* check online table for user */ | /* check online table for user */ | 
 | if (call.LoginACC(&cfg, acc, sess->sess_user, sess->sess_pass) < 1) { | if (call.LoginACC(&cfg, acc, sess->sess_user, sess->sess_pass) < 1) { | 
| ioDEBUG(0, "Login:: DENIED for username %s and password %s", | EVERBOSE(0, "Login:: DENIED for username %s and password %s", | 
 | sess->sess_user, sess->sess_pass); | sess->sess_user, sess->sess_pass); | 
 | ret = MQTT_RETCODE_DENIED; | ret = MQTT_RETCODE_DENIED; | 
 | goto end; | goto end; | 
 | } else { | } else { | 
| ioDEBUG(1, "Login:: ALLOWED for username %s ...", sess->sess_user); | EVERBOSE(1, "Login:: ALLOWED for username %s ...", sess->sess_user); | 
 | ret = MQTT_RETCODE_ACCEPTED; | ret = MQTT_RETCODE_ACCEPTED; | 
 | } | } | 
 |  |  | 
 |  | /* db management */ | 
 | if (call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%") > 0) { | if (call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%") > 0) { | 
| ioDEBUG(2, "Old session %s should be disconnect!", sess->sess_cid); | EVERBOSE(2, "Old session %s should be disconnect!", sess->sess_cid); | 
 | TAILQ_FOREACH(s, &Sessions, sess_node) | TAILQ_FOREACH(s, &Sessions, sess_node) | 
 | if (!strcmp(s->sess_cid, sess->sess_cid)) { | if (!strcmp(s->sess_cid, sess->sess_cid)) { | 
 | /* found stale session & disconnect it! */ | /* found stale session & disconnect it! */ | 
| schedWrite(root, leaveClient, sess, TASK_FD(task), NULL, 0); | sock = s->sess_sock; | 
|  | schedCancelby(root, taskMAX, CRITERIA_FD, (void*) sock, NULL); | 
|  | schedWrite(root, leaveClient, s, s->sess_sock, NULL, 0); | 
 | break; | break; | 
 | } | } | 
 | } | } | 
 |  |  | 
 | if (call.InitSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, sess->sess_addr, | if (call.InitSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, sess->sess_addr, | 
 | sess->sess_will.flag, sess->sess_will.topic, sess->sess_will.msg, | sess->sess_will.flag, sess->sess_will.topic, sess->sess_will.msg, | 
 | sess->sess_will.qos, sess->sess_will.retain) == -1) { | sess->sess_will.qos, sess->sess_will.retain) == -1) { | 
| ioDEBUG(0, "Session %s DENIED for username %s", sess->sess_cid, sess->sess_user); | EVERBOSE(0, "Session %s DENIED for username %s", sess->sess_cid, sess->sess_user); | 
 | ret = MQTT_RETCODE_DENIED; | ret = MQTT_RETCODE_DENIED; | 
 | goto end; | goto end; | 
 | } else { | } else { | 
| ioDEBUG(0, "Session %s from %s and username %s is started", | EVERBOSE(0, "Session %s from %s and username %s is started", | 
 | sess->sess_cid, sess->sess_addr, sess->sess_user); | sess->sess_cid, sess->sess_addr, sess->sess_user); | 
 | ret = MQTT_RETCODE_ACCEPTED; | ret = MQTT_RETCODE_ACCEPTED; | 
 | } | } | 
 |  | /* clean/load session if requested */ | 
 |  | if (sess->sess_clean) { | 
 |  | if (call.DeletePUB_subscribe) | 
 |  | call.DeletePUB_subscribe(&cfg, pub, sess->sess_cid, "%", sess->sess_user, "%"); | 
 |  | if (call.WipePUB_topic) /* delete ALL messages */ | 
 |  | call.WipePUB_topic(&cfg, pub, sess->sess_cid, sess->sess_user, -1); | 
 |  | } else if (call.ReadPUB_subscribe) { | 
 |  | /* load subscribes */ | 
 |  | subs = call.ReadPUB_subscribe(&cfg, pub, sess->sess_cid, "%"); | 
 |  | loadSubscribes(sess, subs); | 
 |  | mqtt_subFree(&subs); | 
 |  | } | 
 |  |  | 
| /* Start session task OK ... */ | /* timer event for retain messages */ | 
| TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); | if (call.ReadPUB_topic) | 
|  | schedTimer(root, sendRetain, sess, ts, NULL, 0); | 
 |  |  | 
 |  | /* Start session task OK ... */ | 
 | if (!schedRead(root, dispatchSession, sess, TASK_FD(task), NULL, 0)) { | if (!schedRead(root, dispatchSession, sess, TASK_FD(task), NULL, 0)) { | 
| ioLIBERR(sched); | ELIBERR(sched); | 
 | ret = MQTT_RETCODE_DENIED; | ret = MQTT_RETCODE_DENIED; | 
| } | } else | 
|  | TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); | 
 |  |  | 
 | call.LOG(logg, "Session %s started from %s for user %s (timeout=%d) OK!\n", sess->sess_cid, | call.LOG(logg, "Session %s started from %s for user %s (timeout=%d) OK!\n", sess->sess_cid, | 
 | sess->sess_addr, sess->sess_user, sess->sess_ka); | sess->sess_addr, sess->sess_user, sess->sess_ka); | 
 | end:    /* close client connection */ | end:    /* close client connection */ | 
 | wlen = mqtt_msgCONNACK(sess->sess_buf, ret); | wlen = mqtt_msgCONNACK(sess->sess_buf, ret); | 
 | if ((wlen = send(TASK_FD(task), sess->sess_buf->msg_base, wlen, 0)) == -1) { | if ((wlen = send(TASK_FD(task), sess->sess_buf->msg_base, wlen, 0)) == -1) { | 
| ioDEBUG(3, "Error:: send(%d) #%d - %s", TASK_FD(task), errno, strerror(errno)); | EVERBOSE(3, "Error:: send(%ld) #%d - %s", (long) TASK_FD(task), errno, strerror(errno)); | 
 | } else | } else | 
| ioDEBUG(5, "Sended %d bytes", wlen); | EVERBOSE(5, "Sended %d bytes", wlen); | 
 |  |  | 
 | if (ret != MQTT_RETCODE_ACCEPTED) { | if (ret != MQTT_RETCODE_ACCEPTED) { | 
| ioDEBUG(1, "Close client %s with socket=%d", sess->sess_addr, TASK_FD(task)); | EVERBOSE(1, "Close client %s with socket=%ld", sess->sess_addr, (long) TASK_FD(task)); | 
 | finiSession(sess); | finiSession(sess); | 
 | } | } | 
 | return NULL; | return NULL; | 
| Line 278  static void * | Line 392  static void * | 
 | acceptClient(sched_task_t *task) | acceptClient(sched_task_t *task) | 
 | { | { | 
 | int cli; | int cli; | 
| io_sockaddr_t sa; | sockaddr_t sa; | 
 | socklen_t sslen = sizeof sa.ss; | socklen_t sslen = sizeof sa.ss; | 
 | ait_val_t *v; | ait_val_t *v; | 
 | char str[STRSIZ]; | char str[STRSIZ]; | 
 |  |  | 
| ioTRACE(4); | ETRACE(4); | 
 |  |  | 
 | assert(task); | assert(task); | 
 |  |  | 
 | if ((cli = accept(TASK_FD(task), &sa.sa, &sslen)) == -1) | if ((cli = accept(TASK_FD(task), &sa.sa, &sslen)) == -1) | 
 | goto end; | goto end; | 
 | else | else | 
| fcntl(cli, F_SETFL, fcntl(cli, F_GETFL, 0) | O_NONBLOCK); | fcntl(cli, F_SETFL, fcntl(cli, F_GETFL) | O_NONBLOCK); | 
 |  |  | 
| v = io_allocVar(); | v = ait_allocVar(); | 
 | if (!v) { | if (!v) { | 
| ioLIBERR(io); | ELIBERR(elwix); | 
 | close(cli); | close(cli); | 
 | goto end; | goto end; | 
 | } else { | } else { | 
 | memset(str, 0, sizeof str); | memset(str, 0, sizeof str); | 
| snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa)); | snprintf(str, sizeof str, "%s:%hu", e_n2addr(&sa, v), e_n2port(&sa)); | 
 | AIT_FREE_VAL(v); | AIT_FREE_VAL(v); | 
 | AIT_SET_STR(v, str); | AIT_SET_STR(v, str); | 
 | } | } | 
| ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v)); | EVERBOSE(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v)); | 
 |  |  | 
 | if (!schedRead(root, startSession, v, cli, NULL, 0)) { | if (!schedRead(root, startSession, v, cli, NULL, 0)) { | 
| io_freeVar(v); | ait_freeVar(&v); | 
 | close(cli); | close(cli); | 
| ioDEBUG(1, "Terminated client with socket=%d", cli); | EVERBOSE(1, "Terminated client with socket=%d", cli); | 
 | } | } | 
 | end: | end: | 
 | if (!schedRead(root, acceptClient, NULL, TASK_FD(task), NULL, 0)) | if (!schedRead(root, acceptClient, NULL, TASK_FD(task), NULL, 0)) | 
| ioLIBERR(sched); | ELIBERR(sched); | 
 | return NULL; | return NULL; | 
 | } | } | 
 |  |  | 
| Line 321  end: | Line 435  end: | 
 | int | int | 
 | Run(int sock) | Run(int sock) | 
 | { | { | 
 | struct tagPub *pub; |  | 
 | struct tagSession *sess; | struct tagSession *sess; | 
 | struct timespec pl = { 0, 100000000 }; | struct timespec pl = { 0, 100000000 }; | 
 |  |  | 
| ioTRACE(1); | ETRACE(1); | 
 |  |  | 
| if (listen(sock, SOMAXCONN) == -1) { | if (mqtt_srv_Listen(sock, 0, 1) == -1) { | 
| ioSYSERR(0); | ELIBERR(mqtt); | 
 | return -1; | return -1; | 
| } else | } | 
| fcntl(sock, F_SETFL, fcntl(sock, F_GETFL, 0) | O_NONBLOCK); |  | 
 |  |  | 
 | /* state machine - accept new connections */ | /* state machine - accept new connections */ | 
 | if (!schedRead(root, acceptClient, NULL, sock, NULL, 0)) { | if (!schedRead(root, acceptClient, NULL, sock, NULL, 0)) { | 
| ioLIBERR(sched); | ELIBERR(sched); | 
 | return -1; | return -1; | 
 | } | } | 
 |  |  | 
 | schedPolling(root, &pl, NULL); | schedPolling(root, &pl, NULL); | 
 | schedRun(root, &Kill); | schedRun(root, &Kill); | 
 |  |  | 
| /* free all undeleted elements into lists */ | schedCancelby(root, taskTIMER, CRITERIA_CALL, sendRetain, NULL); | 
| TAILQ_FOREACH(pub, &Pubs, pub_node) { |  | 
| TAILQ_REMOVE(&Pubs, pub, pub_node); |  | 
 |  |  | 
| AIT_FREE_VAL(&pub->pub_name); | /* free all undeleted elements into lists */ | 
| if (pub->pub_packet.msg_base) |  | 
| free(pub->pub_packet.msg_base); |  | 
| } |  | 
 | TAILQ_FOREACH(sess, &Sessions, sess_node) { | TAILQ_FOREACH(sess, &Sessions, sess_node) { | 
 | TAILQ_REMOVE(&Sessions, sess, sess_node); | TAILQ_REMOVE(&Sessions, sess, sess_node); | 
 |  |  | 
 | finiSession(sess); | finiSession(sess); | 
 | } | } | 
 | return 0; | return 0; |