--- mqtt/src/daemon.c 2011/12/16 02:09:20 1.1.2.23 +++ mqtt/src/daemon.c 2011/12/16 12:16:17 1.1.2.24 @@ -1,5 +1,6 @@ #include "global.h" #include "rtlm.h" +#include "utils.h" #include "mqttd.h" @@ -64,30 +65,42 @@ 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; + + 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); + msg.msg_len = 0; } - pthread_cleanup_push((void(*)(void*)) thrClean, sess); + ioDEBUG(1, "Close socket=%d", sess->sess_sock); + finiSession(sess); +} +static void * +thrSession(struct tagSession *sess) +{ + pthread_cleanup_push((void(*)(void*)) stopSession, sess); + ioTRACE(2); pthread_cleanup_pop(42); @@ -101,7 +114,7 @@ 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; @@ -142,11 +155,16 @@ 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) { - // TODO: disconnect client + 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)) { + 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, @@ -236,7 +254,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; }