--- mqtt/src/daemon.c 2012/04/24 13:54:50 1.2.2.9 +++ mqtt/src/daemon.c 2012/04/27 10:07:50 1.2.2.19 @@ -6,6 +6,7 @@ static void *startSession(sched_task_t *task); +static pthread_attr_t attr; static inline struct tagSession * @@ -22,7 +23,6 @@ initSession(int sock, ait_val_t * __restrict v) sess = malloc(sizeof(struct tagSession)); if (!sess) { ioSYSERR(0); - io_freeVar(v); return NULL; } else memset(sess, 0, sizeof(struct tagSession)); @@ -37,19 +37,10 @@ initSession(int sock, ait_val_t * __restrict v) else sess->sess_retry = strtol(str, NULL, 0); - if (!(sess->sess_root = schedBegin())) { - ioLIBERR(sched); - free(sess); - io_freeVar(v); - return NULL; - } - sess->sess_buf = mqtt_msgAlloc(USHRT_MAX); if (!sess->sess_buf) { ioLIBERR(mqtt); - schedEnd(&sess->sess_root); free(sess); - io_freeVar(v); return NULL; } @@ -58,9 +49,7 @@ initSession(int sock, ait_val_t * __restrict v) if (!sess->sess_srv) { ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError()); mqtt_msgFree(&sess->sess_buf, 42); - schedEnd(&sess->sess_root); free(sess); - io_freeVar(v); return NULL; } else { mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_CONNECT, cmdCONNECT); @@ -74,7 +63,6 @@ initSession(int sock, ait_val_t * __restrict v) sess->sess_sock = sock; strlcpy(sess->sess_addr, (char*) AIT_GET_STR(v), sizeof sess->sess_addr); - io_freeVar(v); return sess; } @@ -92,7 +80,6 @@ finiSession(struct tagSession *sess) call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); SESS_ELEM_LOCK(sess); - while ((store = SLIST_FIRST(&sess->sess_subscr))) { SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node); @@ -106,8 +93,6 @@ finiSession(struct tagSession *sess) SESS_ELEM_UNLOCK(sess); pthread_mutex_destroy(&sess->sess_mtx); - schedEnd(&sess->sess_root); - if (sess->sess_will.msg) free(sess->sess_will.msg); if (sess->sess_will.topic) @@ -147,59 +132,12 @@ stopSession(struct tagSession *sess) sess->sess_addr, sess->sess_user); } -static int -KASession(struct tagSession *sess) -{ - mqtt_msg_t msg = { NULL, 0 }; - 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, MSG_NOSIGNAL)) == -1) { - ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); - 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 * 1000)) == -1 || - pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { - ioDEBUG(3, "Error:: poll(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); - return -1; - } else if (!ret) { - ioDEBUG(5, "Warning:: Session is abandoned ... must be disconnect!"); - return 1; - } - /* receive & decode packet */ - if (recv(sess->sess_sock, 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)); - return -1; - } - if (mqtt_readPINGRESP(sess->sess_buf)) { - ioDEBUG(5, "Warning:: Session is broken, not hear ping response ... must be disconnect!"); - return 2; - } - - /* Keep Alive is OK! */ - return 0; -} - static void * thrSession(struct tagSession *sess) { int ret, locKill = 42; struct pollfd pfd; struct mqtthdr *hdr; - ait_val_t *v; pthread_cleanup_push((void(*)(void*)) stopSession, sess); ioTRACE(2); @@ -211,7 +149,7 @@ thrSession(struct tagSession *sess) pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { ioDEBUG(3, "Error:: poll(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); break; - } else if (!ret && (ret = KASession(sess))) { + } else if (!ret && (ret = mqtt_KeepAlive(sess->sess_sock, sess->sess_ka, 1))) { call.LOG(logg, "Session %s keep-alive missing from %s for user %s ...\n", sess->sess_cid, sess->sess_addr, sess->sess_user); break; @@ -231,20 +169,16 @@ thrSession(struct tagSession *sess) ioLIBERR(mqtt); switch (hdr->mqtt_msg.type) { case MQTT_TYPE_CONNECT: - ioDEBUG(5, "Exec CONNECT session"); - if ((v = io_allocVar())) { - AIT_SET_STR(v, sess->sess_addr); - if (!schedEvent(root, startSession, v, (u_long) sess->sess_sock, sess, ret)) - io_freeVar(v); - } else - ioLIBERR(mqtt); + schedEvent(root, startSession, NULL, (u_long) sess->sess_sock, sess, ret); locKill ^= locKill; - continue; + break; case MQTT_TYPE_DISCONNECT: - ioDEBUG(5, "Exec DISCONNECT session"); finiSession(sess); locKill ^= locKill; continue; + case MQTT_TYPE_SUBSCRIBE: + case MQTT_TYPE_PINGREQ: + break; case MQTT_TYPE_PUBLISH: ioDEBUG(5, "Exec PUBLISH topic QoS=%d", hdr->mqtt_msg.qos); /* @@ -254,13 +188,8 @@ thrSession(struct tagSession *sess) break; case MQTT_TYPE_PUBREL: break; - case MQTT_TYPE_SUBSCRIBE: - break; case MQTT_TYPE_UNSUBSCRIBE: break; - case MQTT_TYPE_PINGREQ: - ioDEBUG(5, "Exec PINGREQ session"); - break; default: ioDEBUG(5, "Error:: Session %s, wrong command %d - DISCARDED", sess->sess_cid, hdr->mqtt_msg.type); @@ -270,6 +199,7 @@ thrSession(struct tagSession *sess) pthread_cleanup_pop(locKill); pthread_exit(NULL); + return NULL; } static void * @@ -281,7 +211,6 @@ startSession(sched_task_t *task) mqtthdr_connack_t cack; struct tagSession *s, *sess = NULL; int ret; - pthread_attr_t attr; ioTRACE(4); @@ -290,8 +219,8 @@ startSession(sched_task_t *task) if (!TASK_DATA(task)) { /* flow from accept new clients */ sess = initSession(TASK_FD(task), TASK_ARG(task)); + io_freeVar(TASK_ARG(task)); if (!sess) { - io_freeVar(TASK_ARG(task)); close(TASK_FD(task)); return NULL; } @@ -333,6 +262,7 @@ startSession(sched_task_t *task) 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) @@ -342,6 +272,7 @@ startSession(sched_task_t *task) 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) { @@ -366,9 +297,6 @@ startSession(sched_task_t *task) } /* Start session thread OK ... */ - pthread_attr_init(&attr); - pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); - SESS_LOCK; TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); pthread_create(&sess->sess_tid, &attr, (void*(*)(void*)) thrSession, sess); @@ -376,8 +304,6 @@ startSession(sched_task_t *task) 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); - - pthread_attr_destroy(&attr); return NULL; end: /* close client connection */ ret = mqtt_msgCONNACK(&msg, ret); @@ -420,6 +346,7 @@ acceptClient(sched_task_t *task) } 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)); @@ -430,7 +357,8 @@ acceptClient(sched_task_t *task) ioDEBUG(1, "Terminated client with socket=%d", cli); } end: - schedRead(TASK_ROOT(task), acceptClient, NULL, TASK_FD(task), NULL, 0); + if (!schedRead(TASK_ROOT(task), acceptClient, NULL, TASK_FD(task), NULL, 0)) + ioLIBERR(sched); return NULL; } @@ -456,8 +384,13 @@ Run(int sock) return -1; } + pthread_attr_init(&attr); + pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); + schedPolling(root, &pl, NULL); schedRun(root, &Kill); + + pthread_attr_destroy(&attr); /* free all undeleted elements into lists */ PUBS_LOCK;