--- mqtt/src/daemon.c 2011/12/09 13:43:55 1.1.2.14 +++ mqtt/src/daemon.c 2011/12/15 12:57:40 1.1.2.21 @@ -1,44 +1,44 @@ #include "global.h" +#include "rtlm.h" #include "mqttd.h" -extern char cliCmd[], *cliStr[]; - - static inline struct tagSession * -initSession(int sock, char * __restrict addr) +initSession(int sock, ait_val_t * __restrict v) { struct tagSession *sess = NULL; const char *str; - if (!addr) + ioTRACE(5); + + if (!v) return NULL; sess = malloc(sizeof(struct tagSession)); if (!sess) { - VERB(3) syslog(LOG_ERR, "Error:: in malloc #%d - %s", errno, strerror(errno)); - free(addr); + ioDEBUG(3, "Error:: in malloc #%d - %s", errno, strerror(errno)); + io_freeVar(v); return NULL; } else memset(sess, 0, sizeof(struct tagSession)); - str = cfg_GetAttribute(&cfg, CFG("mqttd"), CFG("retry")); + str = (const char*) cfg_GetAttribute(&cfg, CFG("mqttd"), CFG("retry")); if (!str) sess->sess_retry = DEFAULT_RETRY; else sess->sess_retry = strtol(str, NULL, 0); - sess->sess_sock = sock; - strlcpy(sess->sess_addr, addr, sizeof sess->sess_addr); - free(addr); - sess->sess_buf = mqtt_msgAlloc(0); if (!sess->sess_buf) { - VERB(3) syslog(LOG_ERR, "Error:: in msgAlloc #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + ioDEBUG(3, "Error:: in msgAlloc #%d - %s", mqtt_GetErrno(), mqtt_GetError()); free(sess); + io_freeVar(v); return NULL; } + sess->sess_sock = sock; + strlcpy(sess->sess_addr, (char*) AIT_GET_STR(v), sizeof sess->sess_addr); + io_freeVar(v); return sess; } @@ -48,6 +48,8 @@ finiSession(struct tagSession *sess) register int i; struct tagStore *store; + ioTRACE(5); + if (!sess) return; @@ -69,39 +71,62 @@ finiSession(struct tagSession *sess) mqtt_msgFree(&sess->sess_buf, 42); - if (sess->sess_tid) - pthread_cancel(sess->sess_tid); free(sess); } static void * +thrSession(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); + } + + pthread_cleanup_push((void(*)(void*)) thrClean, sess); + + ioTRACE(2); + + pthread_cleanup_pop(42); + pthread_exit(NULL); +} + +static 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 msg = { NULL, 0 }, buf = { basebuf, sizeof basebuf }; mqtthdr_connflgs_t flg; mqtthdr_connack_t cack; - int ret = 0; - struct timeval tv = { 0 }; struct tagSession *sess = NULL; + int ret; + pthread_attr_t attr; - FTRACE(4); + ioTRACE(4); sess = initSession(TASK_FD(task), TASK_ARG(task)); - if (!sess) + if (!sess) { + io_freeVar(TASK_ARG(task)); + close(TASK_FD(task)); return NULL; + } + /* receive & decode packet */ if (recv(sess->sess_sock, buf.msg_base, buf.msg_len, 0) == -1) { - VERB(3) syslog(LOG_ERR, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); - goto end; + ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); + finiSession(sess); + return NULL; } 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) { - VERB(3) syslog(LOG_ERR, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + ioDEBUG(3, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError()); goto end; } else { sess->sess_clean = flg.clean_sess; @@ -111,15 +136,52 @@ startSession(sched_task_t *task) } /* 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(0, "Login:: ALLOWED for username %s ...", sess->sess_user); + ret = MQTT_RETCODE_ACCEPTED; + } + ret = mqtt_msgCONNACK(&msg, ret); + 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)); + finiSession(sess); + return NULL; + } else { + ioDEBUG(5, "Sended %d bytes", ret); + free(msg.msg_base); + msg.msg_len = 0; + } + + /* Start session thread OK ... */ + pthread_attr_init(&attr); + pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); + + pthread_mutex_lock(&mtx_sess); + TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); + 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); + + pthread_attr_destroy(&attr); + return NULL; end: /* close client connection */ - if (sess) - free(sess); - 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)); + ret = mqtt_msgCONNACK(&msg, ret); + 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", ret); + free(msg.msg_base); + msg.msg_len = 0; + } + + ioDEBUG(1, "Close client %s with socket=%d", sess->sess_addr, sess->sess_sock); + finiSession(sess); return NULL; } @@ -128,9 +190,15 @@ end: /* close client connection */ static void * thrSched(void *arg __unused) { - FTRACE(1); + struct tagSession *sess; - schedRun(root, (intptr_t*) &Kill); + ioTRACE(1); + + schedRun(root, &Kill); + + TAILQ_FOREACH(sess, &Sessions, sess_node) + if (sess->sess_tid) + pthread_cancel(sess->sess_tid); pthread_exit(NULL); } @@ -140,59 +208,45 @@ Run(int sock) io_sockaddr_t sa; socklen_t sslen = sizeof sa.ss; int cli; - char *str = NULL, szAddr[STRSIZ] = { 0 }; pthread_t tid; + ait_val_t *v; - FTRACE(1); + ioTRACE(1); if (pthread_create(&tid, NULL, thrSched, NULL)) { - syslog(LOG_ERR, "Error:: thread scheduler #%d - %s", errno, strerror(errno)); + ioLOGERR(0); return -1; } else pthread_detach(tid); - VERB(2) syslog(LOG_DEBUG, "Run scheduler management thread"); + ioDEBUG(2, "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); + if (listen(sock, 5 /*SOMAXCONN*/) == -1) { + ioLOGERR(0); return -1; } while (!Kill) { + printf("%%%%%%%%%%rrrrrrrrrrrrrrrrrrrrr errno=%d\n", errno); if ((cli = accept(sock, &sa.sa, &sslen)) == -1) { - syslog(LOG_ERR, "Error:: accept() #%d - %s", errno, strerror(errno)); - continue; + printf("rrrrrrrrrrrrrrrrrrrrrrrrrrrrrrr errno=%d\n", errno); + if (!Kill) + continue; + ioLOGERR(0); + break; + } + v = io_allocVar(); + if (!v) { + ioLIBERR(mqtt); + break; } 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); - } + io_n2addr(&sa, v); + ioDEBUG(1, "Connected client with socket=%d from %s:%d", cli, + AIT_GET_STR(v), io_n2port(&sa)); - if (!schedRead(root, startSession, str, cli)) { + if (!schedRead(root, startSession, v, cli)) { + io_freeVar(v); close(cli); - VERB(1) syslog(LOG_DEBUG, "Terminated client with socket=%d", cli); - if (str) - free(str); + ioDEBUG(1, "Terminated client with socket=%d", cli); } }