--- mqtt/src/daemon.c 2011/12/15 12:58:49 1.1.2.22 +++ mqtt/src/daemon.c 2011/12/16 13:08:46 1.1.2.27 @@ -1,5 +1,6 @@ #include "global.h" #include "rtlm.h" +#include "utils.h" #include "mqttd.h" @@ -64,30 +65,100 @@ finiSession(struct tagSession *sess) if (sess->sess_will.topic) free(sess->sess_will.topic); - if (sess->sess_sock > STDERR_FILENO) { - shutdown(sess->sess_sock, SHUT_RDWR); - close(sess->sess_sock); - } + if (sess->sess_sock > STDERR_FILENO) + srv_Close(sess->sess_sock); mqtt_msgFree(&sess->sess_buf, 42); free(sess); } -static void * -thrSession(struct tagSession *sess) +static void +stopSession(struct tagSession *sess) { - void thrClean(struct tagSession *sess) - { - pthread_mutex_lock(&mtx_sess); - TAILQ_REMOVE(&Sessions, sess, sess_node); - pthread_mutex_unlock(&mtx_sess); - ioDEBUG(1, "Close socket=%d", sess->sess_sock); - finiSession(sess); + mqtt_msg_t msg = { NULL, 0 }; + int ret; + + ioTRACE(4); + + assert(sess); + + pthread_mutex_lock(&mtx_sess); + TAILQ_REMOVE(&Sessions, sess, sess_node); + pthread_mutex_unlock(&mtx_sess); + + ret = mqtt_msgDISCONNECT(&msg); + if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) + ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); + else { + ioDEBUG(5, "Sended %d bytes for disconnect", ret); + free(msg.msg_base); + memset(&msg, 0, sizeof msg); } - pthread_cleanup_push((void(*)(void*)) thrClean, sess); + ioDEBUG(1, "Close socket=%d", sess->sess_sock); + finiSession(sess); + call.LOG(logg, "Session %s stopped from %s for user %s OK!\n", sess->sess_cid, + sess->sess_addr, sess->sess_user); +} + +static int +KASession(struct tagSession *sess) +{ + u_char basebuf[USHRT_MAX]; + mqtt_msg_t msg = { NULL, 0 }, buf = { basebuf, sizeof basebuf }; + int ret; + struct pollfd pfd; + + ioTRACE(4); + + assert(sess); + + /* ping request */ + ret = mqtt_msgPINGREQ(&msg); + if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) { + ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); + stopSession(sess); + return -1; + } else { + ioDEBUG(5, "Sended %d bytes for ping request", ret); + free(msg.msg_base); + memset(&msg, 0, sizeof msg); + } + + pfd.fd = sess->sess_sock; + pfd.events = POLLIN | POLLPRI; + if ((ret = poll(&pfd, 1, sess->sess_ka)) == -1) { + ioDEBUG(3, "Error:: poll(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); + stopSession(sess); + return -1; + } else if (!ret) { + /* problem!!! session is abandoned ... must be disconnect! */ + stopSession(sess); + return 1; + } + /* 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)); + stopSession(sess); + return -1; + } + if (mqtt_readPINGRESP(&buf)) { + /* problem!!! session is broken, not hear ping response ... must be disconnect! */ + stopSession(sess); + return 2; + } + + /* Keep Alive is OK! */ + return 0; +} + +static void * +thrSession(struct tagSession *sess) +{ + pthread_cleanup_push((void(*)(void*)) stopSession, sess); + ioTRACE(2); pthread_cleanup_pop(42); @@ -101,12 +172,14 @@ startSession(sched_task_t *task) mqtt_msg_t msg = { NULL, 0 }, buf = { basebuf, sizeof basebuf }; mqtthdr_connflgs_t flg; mqtthdr_connack_t cack; - struct tagSession *sess = NULL; + struct tagSession *s, *sess = NULL; int ret; pthread_attr_t attr; ioTRACE(4); + assert(task); + sess = initSession(TASK_FD(task), TASK_ARG(task)); if (!sess) { io_freeVar(TASK_ARG(task)); @@ -142,9 +215,29 @@ startSession(sched_task_t *task) ret = MQTT_RETCODE_DENIED; goto end; } else { - ioDEBUG(0, "Login:: ALLOWED for username %s ...", sess->sess_user); + ioDEBUG(1, "Login:: ALLOWED for username %s ...", sess->sess_user); ret = MQTT_RETCODE_ACCEPTED; } + 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! */ + stopSession(s); + 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; + } ret = mqtt_msgCONNACK(&msg, ret); if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) { @@ -154,7 +247,7 @@ startSession(sched_task_t *task) } else { ioDEBUG(5, "Sended %d bytes", ret); free(msg.msg_base); - msg.msg_len = 0; + memset(&msg, 0, sizeof msg); } /* Start session thread OK ... */ @@ -166,7 +259,8 @@ startSession(sched_task_t *task) pthread_create(&sess->sess_tid, &attr, (void*(*)(void*)) thrSession, sess); pthread_mutex_unlock(&mtx_sess); - ioDEBUG(1, "ConnID=%s(%s) for %s login OK!", sess->sess_cid, sess->sess_user, sess->sess_addr); + call.LOG(logg, "Session %s started from %s for user %s OK!\n", sess->sess_cid, + sess->sess_addr, sess->sess_user); pthread_attr_destroy(&attr); return NULL; @@ -177,7 +271,7 @@ end: /* close client connection */ } else { ioDEBUG(5, "Sended %d bytes", ret); free(msg.msg_base); - msg.msg_len = 0; + memset(&msg, 0, sizeof msg); } ioDEBUG(1, "Close client %s with socket=%d", sess->sess_addr, sess->sess_sock); @@ -210,6 +304,7 @@ Run(int sock) int cli; pthread_t tid; ait_val_t *v; + char str[STRSIZ]; ioTRACE(1); @@ -220,7 +315,7 @@ Run(int sock) pthread_detach(tid); ioDEBUG(2, "Run scheduler management thread"); - if (listen(sock, 5 /*SOMAXCONN*/) == -1) { + if (listen(sock, SOMAXCONN) == -1) { ioLOGERR(0); return -1; } @@ -236,10 +331,12 @@ Run(int sock) if (!v) { ioLIBERR(mqtt); break; - } else - io_n2addr(&sa, v); - ioDEBUG(1, "Connected client with socket=%d from %s:%d", cli, - AIT_GET_STR(v), io_n2port(&sa)); + } else { + memset(str, 0, sizeof str); + snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa)); + 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)) { io_freeVar(v);