--- mqtt/src/daemon.c 2011/12/08 17:02:25 1.1.2.11 +++ mqtt/src/daemon.c 2012/08/29 14:40:08 1.4.2.1 @@ -1,151 +1,466 @@ +/************************************************************************* +* (C) 2011 AITNET ltd - Sofia/Bulgaria - +* by Michael Pounov +* +* $Author: misho $ +* $Id: daemon.c,v 1.4.2.1 2012/08/29 14:40:08 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 + 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 "rtlm.h" #include "mqttd.h" +#include "mqttd_calls.h" +#include "utils.h" -extern char cliCmd[], *cliStr[]; +static inline struct tagSession * +initSession(int sock, ait_val_t * __restrict v) +{ + struct tagSession *sess = NULL; + const char *str; + ioTRACE(5); + if (!v) + return NULL; + + sess = io_malloc(sizeof(struct tagSession)); + if (!sess) { + ioSYSERR(0); + return NULL; + } else + memset(sess, 0, sizeof(struct tagSession)); + + SLIST_INIT(&sess->sess_subscr); + + str = cfg_getAttribute(&cfg, "mqttd", "retry"); + if (!str) + sess->sess_retry = DEFAULT_RETRY; + else + sess->sess_retry = strtol(str, NULL, 0); + + sess->sess_buf = mqtt_msgAlloc(USHRT_MAX); + if (!sess->sess_buf) { + ioLIBERR(mqtt); + io_free(sess); + return NULL; + } + + /* init server actor */ + sess->sess_srv = mqtt_srv_cliInit(sock, sess->sess_buf, sess->sess_ka, 1); + if (!sess->sess_srv) { + ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + mqtt_msgFree(&sess->sess_buf, 42); + io_free(sess); + return NULL; + } else { + mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_CONNECT, cmdCONNECT); + mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_PUBLISH, cmdPUBLISH); + mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_PUBREL, cmdPUBREL); + mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_SUBSCRIBE, cmdSUBSCRIBE); + mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_UNSUBSCRIBE, cmdUNSUBSCRIBE); + mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_PINGREQ, cmdPINGREQ); + mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_DISCONNECT, cmdDISCONNECT); + } + + sess->sess_sock = sock; + strlcpy(sess->sess_addr, (char*) AIT_GET_STR(v), sizeof sess->sess_addr); + return sess; +} + +static void +finiSession(struct tagSession *sess) +{ + struct tagStore *store; + + ioTRACE(5); + + if (!sess) + return; + + schedCancelby(root, taskTIMER, CRITERIA_CALL, sendRetain, NULL); + + 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))) { + SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node); + + 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); + } + + if (sess->sess_will.flag) + pubWill(sess); + + if (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) + srv_Close(sess->sess_sock); + + mqtt_srv_cliFini(&sess->sess_srv); + mqtt_msgFree(&sess->sess_buf, 42); + + io_free(sess); +} + static void * +leaveClient(sched_task_t *task) +{ + struct tagSession *sess; + int ret; + + ioTRACE(4); + + assert(task); + + sess = TASK_ARG(task); + assert(sess); + + TAILQ_REMOVE(&Sessions, sess, sess_node); + + ret = mqtt_msgDISCONNECT(sess->sess_buf); + send(TASK_FD(task), sess->sess_buf->msg_base, ret, MSG_NOSIGNAL); + + ioDEBUG(1, "Close socket=%ld", (long) TASK_FD(task)); + call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, + sess->sess_addr, sess->sess_user); + + finiSession(sess); + return NULL; +} + +static void * +dispatchSession(sched_task_t *task) +{ + int ret, len = 0; + struct tagSession *sess; + + ioTRACE(2); + + assert(task); + + sess = TASK_ARG(task); + assert(sess); + + /* receive & decode packet */ + 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)); + TAILQ_REMOVE(&Sessions, sess, sess_node); + finiSession(sess); + return NULL; + } else if (!ret) { + ioDEBUG(4, "Session %s EOF received.", sess->sess_cid); + TAILQ_REMOVE(&Sessions, sess, sess_node); + finiSession(sess); + return NULL; + } + + do { + /* dispatch message type */ + if ((len = mqtt_srv_cliDispatch(sess->sess_srv, ret, sess)) < 0) { + if (len == -1) { + ioLIBERR(mqtt); + 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 + ret -= len; + } while (len > 0 && ret > 0); + + if (len >= 0 && !schedRead(root, dispatchSession, TASK_ARG(task), TASK_FD(task), NULL, 0)) { + ioLIBERR(sched); + TAILQ_REMOVE(&Sessions, sess, sess_node); + finiSession(sess); + } + 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 = io_malloc(sizeof(struct tagStore)); + if (!store) { + ioSYSERR(0); + 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 * startSession(sched_task_t *task) { u_char basebuf[USHRT_MAX]; - mqtt_cb_t cbs[MQTT_TYPE_MAX + 1] = { 0 }; -// mqtt_msg_t *buf = { basebuf, sizeof basebuf }; + mqtt_msg_t buf = { basebuf, sizeof basebuf }; mqtthdr_connflgs_t flg; - int ret = 0; - struct timeval tv = { 0 }; - u_short ka; + mqtthdr_connack_t cack; + ait_val_t *v; + struct tagSession *s, *sess = NULL; + int ret, wlen; + mqtt_subscr_t *subs; + struct timespec ts = { RETAIN_TIMEOUT, 0 }; + intptr_t sock; - FTRACE(4); + ioTRACE(4); - /* - buf = mqtt_msgAlloc(USHRT_MAX); - if (!buf) { - syslog(LOG_ERR, "Error:: allocate message buf (%d) #%d - %s", (int) TASK_FD(task), - mqtt_GetErrno(), mqtt_GetError()); + assert(task); + + if (!TASK_DATA(task)) { + v = TASK_ARG(task); + /* flow from accept new clients */ + sess = initSession(TASK_FD(task), v); + io_freeVar(&v); + if (!sess) { + close(TASK_FD(task)); + return NULL; + } + + /* receive & decode packet */ + 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)); + finiSession(sess); + return NULL; + } + } else { + sess = TASK_DATA(task); + assert(sess); + + buf.msg_len = TASK_DATLEN(task) > sizeof basebuf ? sizeof basebuf : TASK_DATLEN(task); + memcpy(buf.msg_base, sess->sess_buf->msg_base, buf.msg_len); + } + + cack = mqtt_readCONNECT(&buf, &sess->sess_ka, sess->sess_cid, sizeof sess->sess_cid, + sess->sess_user, sizeof sess->sess_user, sess->sess_pass, sizeof sess->sess_pass, + &sess->sess_will.topic, &sess->sess_will.msg); + ret = cack.retcode; + flg.flags = cack.reserved; + if (flg.reserved) { + ioDEBUG(3, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError()); goto end; + } else { + sess->sess_clean = flg.clean_sess; + sess->sess_will.qos = flg.will_qos; + sess->sess_will.retain = flg.will_retain; + sess->sess_will.flag = flg.will_flg; + + sess->sess_srv->timeout = sess->sess_ka; } - */ - if (recv(TASK_FD(task), basebuf, sizeof basebuf, 0) == -1) { - VERB(3) syslog(LOG_ERR, "Error:: recv(%d) #%d - %s", (int) TASK_FD(task), - errno, strerror(errno)); + /* check online table for user */ + if (call.LoginACC(&cfg, acc, sess->sess_user, sess->sess_pass) < 1) { + ioDEBUG(0, "Login:: DENIED for username %s and password %s", + sess->sess_user, sess->sess_pass); + ret = MQTT_RETCODE_DENIED; goto end; + } else { + ioDEBUG(1, "Login:: ALLOWED for username %s ...", sess->sess_user); + ret = MQTT_RETCODE_ACCEPTED; } - /* - flg = mqtt_readCONNECT(buf, &ka, ); - if (flg.reserved) { - VERB(3) syslog(LOG_ERR, "Error:: in MQTT protocol #%d - %s", - mqtt_GetErrno(), mqtt_GetError()); + /* db management */ + if (call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%") > 0) { + ioDEBUG(2, "Old session %s should be disconnect!", sess->sess_cid); + TAILQ_FOREACH(s, &Sessions, sess_node) + if (!strcmp(s->sess_cid, sess->sess_cid)) { + /* found stale session & disconnect it! */ + sock = s->sess_sock; + schedCancelby(root, taskMAX, CRITERIA_FD, (void*) sock, NULL); + schedWrite(root, leaveClient, s, s->sess_sock, NULL, 0); + break; + } + } + 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.qos, sess->sess_will.retain) == -1) { + ioDEBUG(0, "Session %s DENIED for username %s", sess->sess_cid, sess->sess_user); + ret = MQTT_RETCODE_DENIED; goto end; - } */ + } else { + ioDEBUG(0, "Session %s from %s and username %s is started", + sess->sess_cid, sess->sess_addr, sess->sess_user); + 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); + } - /* - for (ret = i = 0; cliCmd[i] && !(ret = (hdr->mqtt_msg.type == cliCmd[i])); i++); - if (!ret) { - VERB(2) syslog(LOG_ERR, "Error:: wrong command type #%d %s", - hdr->mqtt_msg.type, cliStr[i]); - goto end; - } - */ + /* timer event for retain messages */ + if (call.ReadPUB_topic) + schedTimer(root, sendRetain, sess, ts, NULL, 0); - /* check online table for user */ -// ChkSessPUB(&cfg, ); + /* Start session task OK ... */ + if (!schedRead(root, dispatchSession, sess, TASK_FD(task), NULL, 0)) { + ioLIBERR(sched); + ret = MQTT_RETCODE_DENIED; + } else + TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); -// ret = mqttDispatcher(cbs, buf); -// mqtt_msgFree(&buf, 42); + 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); +end: /* close client connection */ + wlen = mqtt_msgCONNACK(sess->sess_buf, ret); + if ((wlen = send(TASK_FD(task), sess->sess_buf->msg_base, wlen, 0)) == -1) { + ioDEBUG(3, "Error:: send(%ld) #%d - %s", (long) TASK_FD(task), errno, strerror(errno)); + } else + ioDEBUG(5, "Sended %d bytes", wlen); - /* - switch (ret) { - case -1: - syslog(LOG_ERR, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError()); - break; - default: - return NULL; + if (ret != MQTT_RETCODE_ACCEPTED) { + ioDEBUG(1, "Close client %s with socket=%ld", sess->sess_addr, (long) TASK_FD(task)); + finiSession(sess); } - */ -end: /* close client connection */ - close(TASK_FD(task)); - VERB(1) syslog(LOG_DEBUG, "Close client %s with socket=%d", - (char*) TASK_ARG(task), (int) TASK_FD(task)); - if (TASK_ARG(task)) - free(TASK_ARG(task)); return NULL; } -/* ----------------------------------------------------------------------- */ - static void * -thrSched(void *arg __unused) +acceptClient(sched_task_t *task) { - FTRACE(1); + int cli; + io_sockaddr_t sa; + socklen_t sslen = sizeof sa.ss; + ait_val_t *v; + char str[STRSIZ]; - schedRun(root, (intptr_t*) &Kill); - pthread_exit(NULL); + ioTRACE(4); + + assert(task); + + if ((cli = accept(TASK_FD(task), &sa.sa, &sslen)) == -1) + goto end; + else + fcntl(cli, F_SETFL, fcntl(cli, F_GETFL) | O_NONBLOCK); + + v = io_allocVar(); + if (!v) { + ioLIBERR(io); + close(cli); + goto end; + } else { + memset(str, 0, sizeof str); + snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa)); + AIT_FREE_VAL(v); + AIT_SET_STR(v, str); + } + ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v)); + + if (!schedRead(root, startSession, v, cli, NULL, 0)) { + io_freeVar(&v); + close(cli); + ioDEBUG(1, "Terminated client with socket=%d", cli); + } +end: + if (!schedRead(root, acceptClient, NULL, TASK_FD(task), NULL, 0)) + ioLIBERR(sched); + return NULL; } +/* ----------------------------------------------------------------------- */ + int Run(int sock) { - io_sockaddr_t sa; - socklen_t sslen = sizeof sa.ss; - int cli; - char *str = NULL, szAddr[STRSIZ] = { 0 }; - pthread_t tid; + struct tagSession *sess; + struct timespec pl = { 0, 100000000 }; - FTRACE(1); + ioTRACE(1); - if (pthread_create(&tid, NULL, thrSched, NULL)) { - syslog(LOG_ERR, "Error:: thread scheduler #%d - %s", errno, strerror(errno)); + if (mqtt_srv_Listen(sock, 0, 1) == -1) { + ioLIBERR(mqtt); return -1; - } else - pthread_detach(tid); - VERB(2) syslog(LOG_DEBUG, "Run scheduler management thread"); + } - if (listen(sock, SOMAXCONN) == -1) { - syslog(LOG_ERR, "Error:: listen(%d) #%d - %s\n", sock, errno, strerror(errno)); - pthread_cancel(tid); + /* state machine - accept new connections */ + if (!schedRead(root, acceptClient, NULL, sock, NULL, 0)) { + ioLIBERR(sched); return -1; } - while (!Kill) { - if ((cli = accept(sock, &sa.sa, &sslen)) == -1) { - syslog(LOG_ERR, "Error:: accept() #%d - %s", errno, strerror(errno)); - continue; - } else - VERB(1) { - switch (sa.sa.sa_family) { - case AF_INET: - inet_ntop(AF_INET, &sa.sin.sin_addr, szAddr, sslen); - snprintf(szAddr, sizeof szAddr, "%s:%d", - szAddr, ntohs(sa.sin.sin_port)); - break; - case AF_INET6: - inet_ntop(AF_INET6, &sa.sin6.sin6_addr, szAddr, sslen); - snprintf(szAddr, sizeof szAddr, "%s:%d", - szAddr, ntohs(sa.sin6.sin6_port)); - break; - case AF_LOCAL: - strlcpy(szAddr, sa.sun.sun_path, sizeof szAddr); - break; - default: - close(cli); - syslog(LOG_ERR, "Error:: unsupported address type %d", - sa.sa.sa_family); - continue; - } - str = strdup(szAddr); - syslog(LOG_DEBUG, "Connected client with socket=%d from %s", cli, str); - } + schedPolling(root, &pl, NULL); + schedRun(root, &Kill); - if (!schedRead(root, startSession, str, cli)) { - close(cli); - VERB(1) syslog(LOG_DEBUG, "Terminated client with socket=%d", cli); - if (str) - free(str); - } - } + schedCancelby(root, taskTIMER, CRITERIA_CALL, sendRetain, NULL); + /* free all undeleted elements into lists */ + TAILQ_FOREACH(sess, &Sessions, sess_node) { + TAILQ_REMOVE(&Sessions, sess, sess_node); + + finiSession(sess); + } return 0; }