version 1.1.2.5, 2011/12/01 08:45:01
|
version 1.1.2.16, 2011/12/12 09:24:25
|
Line 1
|
Line 1
|
#include "global.h" |
#include "global.h" |
|
#include "rtlm.h" |
|
#include "mqttd.h" |
|
|
|
|
|
static inline struct tagSession * |
|
initSession(int sock, char * __restrict addr) |
|
{ |
|
struct tagSession *sess = NULL; |
|
const char *str; |
|
|
|
FTRACE(5); |
|
|
|
if (!addr) |
|
return NULL; |
|
|
|
sess = malloc(sizeof(struct tagSession)); |
|
if (!sess) { |
|
VERB(3) syslog(LOG_ERR, "Error:: in malloc #%d - %s", errno, strerror(errno)); |
|
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_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; |
|
} |
|
|
|
sess->sess_sock = sock; |
|
strlcpy(sess->sess_addr, addr, sizeof sess->sess_addr); |
|
free(addr); |
|
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); |
|
|
|
if (sess->sess_tid) |
|
pthread_cancel(sess->sess_tid); |
|
free(sess); |
|
} |
|
|
static void * |
static void * |
|
thrSession(struct tagSession *sess) |
|
{ |
|
FTRACE(2); |
|
|
|
TAILQ_REMOVE(&Sessions, sess, sess_node); |
|
pthread_exit(NULL); |
|
return NULL; |
|
} |
|
|
|
static void * |
startSession(sched_task_t *task) |
startSession(sched_task_t *task) |
{ |
{ |
close(TASK_FD(task)); | u_char basebuf[USHRT_MAX]; |
| mqtt_msg_t buf = { basebuf, sizeof basebuf }; |
| mqtthdr_connflgs_t flg; |
| mqtthdr_connack_t cack; |
| struct tagSession *sess = NULL; |
| |
| FTRACE(4); |
| |
| sess = initSession(TASK_FD(task), TASK_ARG(task)); |
| if (!sess) { |
| if (TASK_ARG(task)) |
| free(TASK_ARG(task)); |
| close(TASK_FD(task)); |
| return NULL; |
| } |
| |
| 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; |
| } |
| 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); |
| 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)) { |
| 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); |
| |
| /* Start session thread OK ... */ |
| TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); |
| pthread_create(&sess->sess_tid, NULL, (void*(*)(void*)) thrSession, sess); |
| pthread_detach(sess->sess_tid); |
| VERB(1) syslog(LOG_DEBUG, "ConnID=%s(%s) for %s login OK!", |
| sess->sess_cid, sess->sess_user, sess->sess_addr); |
| return NULL; |
| end: /* close client connection */ |
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)); |
if (TASK_ARG(task)) | finiSession(sess); |
free(TASK_ARG(task)); | return NULL; |
return 0; | |
} |
} |
|
|
/* ----------------------------------------------------------------------- */ |
/* ----------------------------------------------------------------------- */ |