version 1.1.2.7, 2011/12/05 08:15:50
|
version 1.1.2.14, 2011/12/09 13:43:55
|
Line 1
|
Line 1
|
#include "global.h" |
#include "global.h" |
|
#include "mqttd.h" |
|
|
|
|
extern char cliCmd[], *cliStr[]; |
extern char cliCmd[], *cliStr[]; |
|
|
|
|
|
static inline struct tagSession * |
|
initSession(int sock, char * __restrict addr) |
|
{ |
|
struct tagSession *sess = NULL; |
|
const char *str; |
|
|
|
if (!addr) |
|
return NULL; |
|
|
|
sess = malloc(sizeof(struct tagSession)); |
|
if (!sess) { |
|
VERB(3) syslog(LOG_ERR, "Error:: in malloc #%d - %s", errno, strerror(errno)); |
|
free(addr); |
|
return NULL; |
|
} else |
|
memset(sess, 0, sizeof(struct tagSession)); |
|
|
|
str = 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()); |
|
free(sess); |
|
return NULL; |
|
} |
|
|
|
return sess; |
|
} |
|
|
|
static void |
|
finiSession(struct tagSession *sess) |
|
{ |
|
register int i; |
|
struct tagStore *store; |
|
|
|
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); |
|
|
|
if (sess->sess_tid) |
|
pthread_cancel(sess->sess_tid); |
|
free(sess); |
|
} |
|
|
static void * |
static void * |
startSession(sched_task_t *task) |
startSession(sched_task_t *task) |
{ |
{ |
mqtt_msg_t *buf; | u_char basebuf[USHRT_MAX]; |
mqtt_cb_t cbs[MQTT_TYPE_MAX + 1] = { 0 }; |
mqtt_cb_t cbs[MQTT_TYPE_MAX + 1] = { 0 }; |
struct mqtthdr *hdr; | mqtt_msg_t buf = { basebuf, sizeof basebuf }; |
| mqtthdr_connflgs_t flg; |
| mqtthdr_connack_t cack; |
int ret = 0; |
int ret = 0; |
register int i; | struct timeval tv = { 0 }; |
| struct tagSession *sess = NULL; |
|
|
FTRACE(4); |
FTRACE(4); |
|
|
buf = mqtt_msgAlloc(USHRT_MAX); | sess = initSession(TASK_FD(task), TASK_ARG(task)); |
if (!buf) { | if (!sess) |
syslog(LOG_ERR, "Error:: allocate message buf (%d) #%d - %s", (int) TASK_FD(task), | return NULL; |
mqtt_GetErrno(), mqtt_GetError()); | |
| 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; |
goto end; |
} |
} |
| cack = mqtt_readCONNECT(&buf, &sess->sess_ka, sess->sess_cid, sizeof sess->sess_cid, |
if (recv(TASK_FD(task), buf->msg_base, buf->msg_len, 0) == -1) { | sess->sess_user, sizeof sess->sess_user, sess->sess_pass, sizeof sess->sess_pass, |
syslog(LOG_ERR, "Error:: recv(%d) #%d - %s", (int) TASK_FD(task), | &sess->sess_will.topic, &sess->sess_will.msg); |
errno, strerror(errno)); | flg.flags = cack.reserved; |
mqtt_msgFree(&buf, 42); | if (flg.reserved) { |
| VERB(3) syslog(LOG_ERR, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
goto end; |
goto end; |
} else { |
} else { |
hdr = (struct mqtthdr*) buf->msg_base; | sess->sess_clean = flg.clean_sess; |
for (ret = i = 0; cliCmd[i] && !(ret = (hdr->mqtt_msg.type == cliCmd[i])); i++); | sess->sess_will.qos = flg.will_qos; |
if (!ret) { | sess->sess_will.retain = flg.will_retain; |
syslog(LOG_ERR, "Error:: wrong command type #%d %s", | sess->sess_will.flag = flg.will_flg; |
hdr->mqtt_msg.type, cliStr[i]); | |
mqtt_msgFree(&buf, 42); | |
goto end; | |
} | |
} |
} |
|
|
ret = mqttDispatcher(cbs, buf); | /* check online table for user */ |
|
|
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 */ |
end: /* close client connection */ |
|
if (sess) |
|
free(sess); |
close(TASK_FD(task)); |
close(TASK_FD(task)); |
VERB(1) syslog(LOG_DEBUG, "Close client %s with socket=%d", |
VERB(1) syslog(LOG_DEBUG, "Close client %s with socket=%d", |
(char*) TASK_ARG(task), (int) TASK_FD(task)); |
(char*) TASK_ARG(task), (int) TASK_FD(task)); |