--- mqtt/src/daemon.c 2011/12/05 10:37:17 1.1.2.8 +++ mqtt/src/daemon.c 2011/12/09 11:46:48 1.1.2.13 @@ -1,4 +1,5 @@ #include "global.h" +#include "mqttd.h" extern char cliCmd[], *cliStr[]; @@ -7,59 +8,42 @@ extern char cliCmd[], *cliStr[]; static void * startSession(sched_task_t *task) { - static u_char basebuf[USHRT_MAX]; - mqtt_msg_t *buf; + u_char basebuf[USHRT_MAX]; mqtt_cb_t cbs[MQTT_TYPE_MAX + 1] = { 0 }; - struct mqtthdr *hdr; - int len, ret = 0; + mqtt_msg_t buf = { basebuf, sizeof basebuf }; + mqtthdr_connflgs_t flg; + mqtthdr_connack_t cack; + int ret = 0; + struct timeval tv = { 0 }; + u_short ka; + struct tagSession *sess = NULL; FTRACE(4); - /* - buf = mqtt_msgAlloc(USHRT_MAX); - if (!buf) { - syslog(LOG_ERR, "Error:: allocate message buf (%d) #%d - %s", (int) TASK_FD(task), - mqtt_GetErrno(), mqtt_GetError()); + sess = malloc(sizeof(struct tagSession)); + if (!sess) { + VERB(3) syslog(LOG_ERR, "Error:: in malloc #%d - %s", errno, strerror(errno)); goto end; } - */ - if (recv(TASK_FD(task), basebuf, sizeof basebuf, 0) == -1) { + if (recv(TASK_FD(task), buf.msg_base, buf.msg_len, 0) == -1) { VERB(3) syslog(LOG_ERR, "Error:: recv(%d) #%d - %s", (int) TASK_FD(task), errno, strerror(errno)); goto end; - } else - hdr = (struct mqtthdr*) basebuf; - if (hdr->mqtt_msg.type != MQTT_TYPE_CONNECT) { - VERB(2) syslog(LOG_ERR, "Error:: wrong command type for 1st command #%d", - hdr->mqtt_msg.type); + } + cack = mqtt_readCONNECT(&buf, &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); + if (cack.reserved) { + VERB(3) syslog(LOG_ERR, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError()); goto end; - } else - len = mqtt_decodeLen(hdr->mqtt_len, &ret); + } - /* - for (ret = i = 0; cliCmd[i] && !(ret = (hdr->mqtt_msg.type == cliCmd[i])); i++); - if (!ret) { - VERB(2) syslog(LOG_ERR, "Error:: wrong command type #%d %s", - hdr->mqtt_msg.type, cliStr[i]); - goto end; - } - */ - /* check online table for user */ -// ChkSessPUB(&cfg, ); - ret = mqttDispatcher(cbs, buf); - mqtt_msgFree(&buf, 42); - - switch (ret) { - case -1: - syslog(LOG_ERR, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError()); - break; - default: - 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));