|
version 1.1.2.23, 2011/12/16 02:09:20
|
version 1.1.2.24, 2011/12/16 12:16:17
|
|
Line 1
|
Line 1
|
| #include "global.h" |
#include "global.h" |
| #include "rtlm.h" |
#include "rtlm.h" |
| |
#include "utils.h" |
| #include "mqttd.h" |
#include "mqttd.h" |
| |
|
| |
|
|
Line 64 finiSession(struct tagSession *sess)
|
Line 65 finiSession(struct tagSession *sess)
|
| if (sess->sess_will.topic) |
if (sess->sess_will.topic) |
| free(sess->sess_will.topic); |
free(sess->sess_will.topic); |
| |
|
| if (sess->sess_sock > STDERR_FILENO) { | if (sess->sess_sock > STDERR_FILENO) |
| shutdown(sess->sess_sock, SHUT_RDWR); | srv_Close(sess->sess_sock); |
| close(sess->sess_sock); | |
| } | |
| |
|
| mqtt_msgFree(&sess->sess_buf, 42); |
mqtt_msgFree(&sess->sess_buf, 42); |
| |
|
| free(sess); |
free(sess); |
| } |
} |
| |
|
| static void * | static void |
| thrSession(struct tagSession *sess) | stopSession(struct tagSession *sess) |
| { |
{ |
| void thrClean(struct tagSession *sess) | mqtt_msg_t msg = { NULL, 0 }; |
| { | int ret; |
| pthread_mutex_lock(&mtx_sess); | |
| TAILQ_REMOVE(&Sessions, sess, sess_node); | pthread_mutex_lock(&mtx_sess); |
| pthread_mutex_unlock(&mtx_sess); | TAILQ_REMOVE(&Sessions, sess, sess_node); |
| ioDEBUG(1, "Close socket=%d", sess->sess_sock); | pthread_mutex_unlock(&mtx_sess); |
| finiSession(sess); | |
| | ret = mqtt_msgDISCONNECT(&msg); |
| | if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) |
| | ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); |
| | else { |
| | ioDEBUG(5, "Sended %d bytes for disconnect", ret); |
| | free(msg.msg_base); |
| | msg.msg_len = 0; |
| } |
} |
| |
|
| pthread_cleanup_push((void(*)(void*)) thrClean, sess); | ioDEBUG(1, "Close socket=%d", sess->sess_sock); |
| | finiSession(sess); |
| | } |
| |
|
| |
static void * |
| |
thrSession(struct tagSession *sess) |
| |
{ |
| |
pthread_cleanup_push((void(*)(void*)) stopSession, sess); |
| |
|
| ioTRACE(2); |
ioTRACE(2); |
| |
|
| pthread_cleanup_pop(42); |
pthread_cleanup_pop(42); |
|
Line 101 startSession(sched_task_t *task)
|
Line 114 startSession(sched_task_t *task)
|
| mqtt_msg_t msg = { NULL, 0 }, buf = { basebuf, sizeof basebuf }; |
mqtt_msg_t msg = { NULL, 0 }, buf = { basebuf, sizeof basebuf }; |
| mqtthdr_connflgs_t flg; |
mqtthdr_connflgs_t flg; |
| mqtthdr_connack_t cack; |
mqtthdr_connack_t cack; |
| struct tagSession *sess = NULL; | struct tagSession *s, *sess = NULL; |
| int ret; |
int ret; |
| pthread_attr_t attr; |
pthread_attr_t attr; |
| |
|
|
Line 142 startSession(sched_task_t *task)
|
Line 155 startSession(sched_task_t *task)
|
| ret = MQTT_RETCODE_DENIED; |
ret = MQTT_RETCODE_DENIED; |
| goto end; |
goto end; |
| } else { |
} else { |
| ioDEBUG(0, "Login:: ALLOWED for username %s ...", sess->sess_user); | ioDEBUG(1, "Login:: ALLOWED for username %s ...", sess->sess_user); |
| ret = MQTT_RETCODE_ACCEPTED; |
ret = MQTT_RETCODE_ACCEPTED; |
| } |
} |
| if (call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%") > 0) { |
if (call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%") > 0) { |
| // TODO: disconnect client | ioDEBUG(2, "Old session %s should be disconnect!", sess->sess_cid); |
| | TAILQ_FOREACH(s, &Sessions, sess_node) |
| | if (!strcmp(s->sess_cid, sess->sess_cid)) { |
| | stopSession(s); |
| | break; |
| | } |
| } |
} |
| if (call.InitSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, sess->sess_addr, |
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.flag, sess->sess_will.topic, sess->sess_will.msg, |
|
Line 236 Run(int sock)
|
Line 254 Run(int sock)
|
| pthread_detach(tid); |
pthread_detach(tid); |
| ioDEBUG(2, "Run scheduler management thread"); |
ioDEBUG(2, "Run scheduler management thread"); |
| |
|
| if (listen(sock, 5 /*SOMAXCONN*/) == -1) { | if (listen(sock, SOMAXCONN) == -1) { |
| ioLOGERR(0); |
ioLOGERR(0); |
| return -1; |
return -1; |
| } |
} |