--- mqtt/src/daemon.c 2011/11/30 00:14:42 1.1.2.1 +++ mqtt/src/daemon.c 2011/12/14 12:45:30 1.1.2.18 @@ -1,3 +1,246 @@ #include "global.h" +#include "rtlm.h" +#include "mqttd.h" +static inline struct tagSession * +initSession(int sock, ait_val_t * __restrict v) +{ + struct tagSession *sess = NULL; + const char *str; + + FTRACE(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)); + io_freeVar(v); + return NULL; + } else + memset(sess, 0, sizeof(struct tagSession)); + + 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_buf = mqtt_msgAlloc(0); + if (!sess->sess_buf) { + VERB(3) syslog(LOG_ERR, "Error:: in msgAlloc #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + free(sess); + io_freeVar(v); + return NULL; + } + + sess->sess_sock = sock; + strlcpy(sess->sess_addr, AIT_GET_STR(v), sizeof sess->sess_addr); + io_freeVar(v); + return sess; +} + +static void +finiSession(struct tagSession *sess) +{ + register int i; + struct tagStore *store; + + FTRACE(5); + + if (!sess) + return; + + for (i = 0; i < MQTT_QOS_RESERVED; i++) + while ((store = SLIST_FIRST(&sess->sess_store[i]))) { + SLIST_REMOVE_HEAD(&sess->sess_store[i], st_node); + free(store); + } + + if (sess->sess_will.msg) + free(sess->sess_will.msg); + 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); + } + + mqtt_msgFree(&sess->sess_buf, 42); + + 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); + VERB(1) syslog(LOG_DEBUG, "Close socket=%d", sess->sess_sock); + finiSession(sess); + } + + pthread_cleanup_push((void(*)(void*)) thrClean, sess); + + FTRACE(2); + + pthread_cleanup_pop(42); + pthread_exit(NULL); +} + +static void * +startSession(sched_task_t *task) +{ + u_char basebuf[USHRT_MAX]; + mqtt_msg_t msg = { NULL, 0 }, buf = { basebuf, sizeof basebuf }; + mqtthdr_connflgs_t flg; + mqtthdr_connack_t cack; + struct tagSession *sess = NULL; + int ret; + pthread_attr_t attr; + + FTRACE(4); + + sess = initSession(TASK_FD(task), TASK_ARG(task)); + 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)); + 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()); + 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; + } + + /* check online table for user */ + if (call.LoginACC(&cfg, acc, sess->sess_user, sess->sess_pass) < 1) { + VERB(1) syslog(LOG_WARNING, "Login:: DENIED for username %s and password %s", + sess->sess_user, sess->sess_pass); + goto end; + } else + VERB(1) syslog(LOG_WARNING, "Login:: ALLOWED for username %s ...", sess->sess_user); + + ret = mqtt_msgCONNACK(&msg, ret); + if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) { + VERB(3) syslog(LOG_ERR, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); + finiSession(sess); + return NULL; + } else { + VERB(5) ioLOGGER(LOG_DEBUG, "Sended %d bytes", ret); + mqtt_msgFree((mqtt_msg_t**) &msg, 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); + + VERB(1) ioLOGGER(LOG_DEBUG, "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 */ + ret = mqtt_msgCONNACK(&msg, ret); + if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) { + VERB(3) syslog(LOG_ERR, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); + } else { + VERB(5) ioLOGGER(LOG_DEBUG, "Sended %d bytes", ret); + mqtt_msgFree((mqtt_msg_t**) &msg, 0); + } + + VERB(1) ioLOGGER(LOG_DEBUG, "Close client %s with socket=%d", sess->sess_addr, sess->sess_sock); + finiSession(sess); + return NULL; +} + +/* ----------------------------------------------------------------------- */ + +static void * +thrSched(void *arg __unused) +{ + FTRACE(1); + struct tagSession *sess; + + schedRun(root, (intptr_t*) &Kill); + + TAILQ_FOREACH(sess, &Sessions, sess_node) + if (sess->sess_tid) + pthread_cancel(sess->sess_tid); + pthread_exit(NULL); +} + +int +Run(int sock) +{ + io_sockaddr_t sa; + socklen_t sslen = sizeof sa.ss; + int cli; + pthread_t tid; + ait_val_t *v; + + FTRACE(1); + + if (pthread_create(&tid, NULL, thrSched, NULL)) { + syslog(LOG_ERR, "Error:: thread scheduler #%d - %s", errno, strerror(errno)); + return -1; + } else + pthread_detach(tid); + VERB(2) ioLOGGER(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); + return -1; + } + + while (!Kill) { + if ((cli = accept(sock, &sa.sa, &sslen)) == -1) { + syslog(LOG_ERR, "Error:: accept() #%d - %s", errno, strerror(errno)); + continue; + } + v = io_allocVar(); + if (!v) { + ioLIBERR(mqtt); + break; + } else + io_n2addr(&sa, v); + VERB(1) ioLOGGER(LOG_DEBUG, "Connected client with socket=%d from %s:%d", cli, + AIT_GET_STR(v), io_n2port(&sa)); + + if (!schedRead(root, startSession, v, cli)) { + io_freeVar(v); + close(cli); + VERB(1) ioLOGGER(LOG_DEBUG, "Terminated client with socket=%d", cli); + } + } + + return 0; +}