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; |
} |
} |