version 1.2.2.11, 2012/04/25 12:04:30
|
version 1.2.2.26, 2012/05/05 15:04:29
|
Line 5
|
Line 5
|
#include "mqttd_calls.h" |
#include "mqttd_calls.h" |
|
|
|
|
static void *startSession(sched_task_t *task); |
|
static pthread_attr_t attr; |
|
|
|
|
|
static inline struct tagSession * |
static inline struct tagSession * |
initSession(int sock, ait_val_t * __restrict v) |
initSession(int sock, ait_val_t * __restrict v) |
{ |
{ |
Line 23 initSession(int sock, ait_val_t * __restrict v)
|
Line 19 initSession(int sock, ait_val_t * __restrict v)
|
sess = malloc(sizeof(struct tagSession)); |
sess = malloc(sizeof(struct tagSession)); |
if (!sess) { |
if (!sess) { |
ioSYSERR(0); |
ioSYSERR(0); |
io_freeVar(v); |
|
return NULL; |
return NULL; |
} else |
} else |
memset(sess, 0, sizeof(struct tagSession)); |
memset(sess, 0, sizeof(struct tagSession)); |
|
|
pthread_mutex_init(&sess->sess_mtx, NULL); |
|
|
|
SLIST_INIT(&sess->sess_subscr); |
SLIST_INIT(&sess->sess_subscr); |
|
|
str = cfg_getAttribute(&cfg, "mqttd", "retry"); |
str = cfg_getAttribute(&cfg, "mqttd", "retry"); |
Line 38 initSession(int sock, ait_val_t * __restrict v)
|
Line 31 initSession(int sock, ait_val_t * __restrict v)
|
else |
else |
sess->sess_retry = strtol(str, NULL, 0); |
sess->sess_retry = strtol(str, NULL, 0); |
|
|
if (!(sess->sess_root = schedBegin())) { |
|
ioLIBERR(sched); |
|
free(sess); |
|
io_freeVar(v); |
|
return NULL; |
|
} |
|
|
|
sess->sess_buf = mqtt_msgAlloc(USHRT_MAX); |
sess->sess_buf = mqtt_msgAlloc(USHRT_MAX); |
if (!sess->sess_buf) { |
if (!sess->sess_buf) { |
ioLIBERR(mqtt); |
ioLIBERR(mqtt); |
schedEnd(&sess->sess_root); |
|
free(sess); |
free(sess); |
io_freeVar(v); |
|
return NULL; |
return NULL; |
} |
} |
|
|
Line 59 initSession(int sock, ait_val_t * __restrict v)
|
Line 43 initSession(int sock, ait_val_t * __restrict v)
|
if (!sess->sess_srv) { |
if (!sess->sess_srv) { |
ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
mqtt_msgFree(&sess->sess_buf, 42); |
mqtt_msgFree(&sess->sess_buf, 42); |
schedEnd(&sess->sess_root); |
|
free(sess); |
free(sess); |
io_freeVar(v); |
|
return NULL; |
return NULL; |
} else { |
} else { |
mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_CONNECT, cmdCONNECT); |
mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_CONNECT, cmdCONNECT); |
Line 75 initSession(int sock, ait_val_t * __restrict v)
|
Line 57 initSession(int sock, ait_val_t * __restrict v)
|
|
|
sess->sess_sock = sock; |
sess->sess_sock = sock; |
strlcpy(sess->sess_addr, (char*) AIT_GET_STR(v), sizeof sess->sess_addr); |
strlcpy(sess->sess_addr, (char*) AIT_GET_STR(v), sizeof sess->sess_addr); |
io_freeVar(v); |
|
return sess; |
return sess; |
} |
} |
|
|
Line 92 finiSession(struct tagSession *sess)
|
Line 73 finiSession(struct tagSession *sess)
|
if (call.FiniSessPUB) |
if (call.FiniSessPUB) |
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); |
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); |
|
|
SESS_ELEM_LOCK(sess); |
|
|
|
while ((store = SLIST_FIRST(&sess->sess_subscr))) { |
while ((store = SLIST_FIRST(&sess->sess_subscr))) { |
SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node); |
SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node); |
|
|
Line 104 finiSession(struct tagSession *sess)
|
Line 83 finiSession(struct tagSession *sess)
|
|
|
free(store); |
free(store); |
} |
} |
SESS_ELEM_UNLOCK(sess); |
|
pthread_mutex_destroy(&sess->sess_mtx); |
|
|
|
schedEnd(&sess->sess_root); |
|
|
|
if (sess->sess_will.msg) |
if (sess->sess_will.msg) |
free(sess->sess_will.msg); |
free(sess->sess_will.msg); |
if (sess->sess_will.topic) |
if (sess->sess_will.topic) |
Line 123 finiSession(struct tagSession *sess)
|
Line 98 finiSession(struct tagSession *sess)
|
free(sess); |
free(sess); |
} |
} |
|
|
static void | static void * |
stopSession(struct tagSession *sess) | leaveClient(sched_task_t *task) |
{ |
{ |
mqtt_msg_t msg = { NULL, 0 }; | struct tagSession *sess; |
int ret; |
int ret; |
|
|
ioTRACE(4); |
ioTRACE(4); |
|
|
|
assert(task); |
|
|
|
sess = TASK_ARG(task); |
assert(sess); |
assert(sess); |
|
|
SESS_LOCK; |
|
TAILQ_REMOVE(&Sessions, sess, sess_node); |
TAILQ_REMOVE(&Sessions, sess, sess_node); |
SESS_UNLOCK; |
|
|
|
ret = mqtt_msgDISCONNECT(&msg); | ret = mqtt_msgDISCONNECT(sess->sess_buf); |
send(sess->sess_sock, msg.msg_base, ret, MSG_NOSIGNAL); | send(TASK_FD(task), sess->sess_buf->msg_base, ret, MSG_NOSIGNAL); |
free(msg.msg_base); | |
|
|
ioDEBUG(1, "Close socket=%d", sess->sess_sock); | ioDEBUG(1, "Close socket=%d", TASK_FD(task)); |
finiSession(sess); | |
| |
call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, |
call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, |
sess->sess_addr, sess->sess_user); |
sess->sess_addr, sess->sess_user); |
|
|
|
finiSession(sess); |
|
return NULL; |
} |
} |
|
|
static int | static void * |
KASession(struct tagSession *sess) | dispatchSession(sched_task_t *task) |
{ |
{ |
mqtt_msg_t msg = { NULL, 0 }; | int ret, len = 0; |
int ret; | struct tagSession *sess; |
struct pollfd pfd; | |
|
|
ioTRACE(4); | ioTRACE(2); |
|
|
|
assert(task); |
|
|
|
sess = TASK_ARG(task); |
assert(sess); |
assert(sess); |
|
|
/* ping request */ |
|
ret = mqtt_msgPINGREQ(&msg); |
|
if ((ret = send(sess->sess_sock, msg.msg_base, ret, MSG_NOSIGNAL)) == -1) { |
|
ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); |
|
return -1; |
|
} else { |
|
ioDEBUG(5, "Sended %d bytes for ping request", ret); |
|
free(msg.msg_base); |
|
memset(&msg, 0, sizeof msg); |
|
} |
|
|
|
pfd.fd = sess->sess_sock; |
|
pfd.events = POLLIN | POLLPRI; |
|
if ((ret = poll(&pfd, 1, sess->sess_ka * 1000)) == -1 || |
|
pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { |
|
ioDEBUG(3, "Error:: poll(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); |
|
return -1; |
|
} else if (!ret) { |
|
ioDEBUG(5, "Warning:: Session is abandoned ... must be disconnect!"); |
|
return 1; |
|
} |
|
/* receive & decode packet */ |
/* receive & decode packet */ |
if (recv(sess->sess_sock, sess->sess_buf->msg_base, sess->sess_buf->msg_len, 0) == -1) { | if ((ret = recv(TASK_FD(task), sess->sess_buf->msg_base, sess->sess_buf->msg_len, 0)) == -1) { |
ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); |
ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); |
return -1; | TAILQ_REMOVE(&Sessions, sess, sess_node); |
| finiSession(sess); |
| return NULL; |
| } else if (!ret) { |
| ioDEBUG(4, "Session %s EOF received.", sess->sess_cid); |
| TAILQ_REMOVE(&Sessions, sess, sess_node); |
| finiSession(sess); |
| return NULL; |
} |
} |
if (mqtt_readPINGRESP(sess->sess_buf)) { |
|
ioDEBUG(5, "Warning:: Session is broken, not hear ping response ... must be disconnect!"); |
|
return 2; |
|
} |
|
|
|
/* Keep Alive is OK! */ | do { |
return 0; | /* dispatch message type */ |
} | if ((len = mqtt_srv_Dispatch(sess->sess_srv, ret, sess)) < 0) { |
| if (len == -1) |
static void * | ioLIBERR(mqtt); |
thrSession(struct tagSession *sess) | TAILQ_REMOVE(&Sessions, sess, sess_node); |
{ | finiSession(sess); |
int ret, locKill = 42; | |
struct pollfd pfd; | |
struct mqtthdr *hdr; | |
ait_val_t *v; | |
| |
pthread_cleanup_push((void(*)(void*)) stopSession, sess); | |
ioTRACE(2); | |
| |
pfd.fd = sess->sess_sock; | |
pfd.events = POLLIN | POLLPRI; | |
while (!Kill && locKill) { | |
if ((ret = poll(&pfd, 1, sess->sess_ka * 1000)) == -1 || | |
pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { | |
ioDEBUG(3, "Error:: poll(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); | |
break; | |
} else if (!ret && (ret = KASession(sess))) { | |
call.LOG(logg, "Session %s keep-alive missing from %s for user %s ...\n", | |
sess->sess_cid, sess->sess_addr, sess->sess_user); | |
break; | |
} | |
/* receive & decode packet */ | |
if ((ret = recv(sess->sess_sock, sess->sess_buf->msg_base, sess->sess_buf->msg_len, 0)) == -1) { | |
ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); | |
break; | |
} else if (!ret) { | |
ioDEBUG(4, "Session %s EOF received.", sess->sess_cid); | |
break; | |
} else |
} else |
hdr = (struct mqtthdr*) sess->sess_buf->msg_base; | ret -= len; |
| } while (len > 0 && ret > 0); |
|
|
/* dispatch message type */ | if (len >= 0 && !schedRead(root, dispatchSession, TASK_ARG(task), TASK_FD(task), NULL, 0)) { |
if (mqtt_srv_Dispatch(sess->sess_srv, sess)) | ioLIBERR(sched); |
ioLIBERR(mqtt); | TAILQ_REMOVE(&Sessions, sess, sess_node); |
locKill ^= locKill; | finiSession(sess); |
switch (hdr->mqtt_msg.type) { | |
case MQTT_TYPE_CONNECT: | |
ioDEBUG(5, "Exec CONNECT session"); | |
if ((v = io_allocVar())) { | |
AIT_SET_STR(v, sess->sess_addr); | |
if (!schedEvent(root, startSession, v, (u_long) sess->sess_sock, sess, ret)) | |
io_freeVar(v); | |
} else | |
ioLIBERR(mqtt); | |
continue; | |
case MQTT_TYPE_DISCONNECT: | |
ioDEBUG(5, "Exec DISCONNECT session"); | |
finiSession(sess); | |
locKill = 42; | |
continue; | |
case MQTT_TYPE_PUBLISH: | |
ioDEBUG(5, "Exec PUBLISH topic QoS=%d", hdr->mqtt_msg.qos); | |
/* | |
if (cmdPUBLISH(sess)) | |
locKill ^= locKill; | |
*/ | |
break; | |
case MQTT_TYPE_PUBREL: | |
break; | |
case MQTT_TYPE_SUBSCRIBE: | |
break; | |
case MQTT_TYPE_UNSUBSCRIBE: | |
break; | |
case MQTT_TYPE_PINGREQ: | |
ioDEBUG(5, "Exec PINGREQ session"); | |
break; | |
case MQTT_TYPE_PINGRESP: | |
ioDEBUG(5, "Exec PINGRESP session"); | |
break; | |
default: | |
ioDEBUG(5, "Error:: Session %s, wrong command %d - DISCARDED", | |
sess->sess_cid, hdr->mqtt_msg.type); | |
break; | |
} | |
} |
} |
|
|
pthread_cleanup_pop(locKill); |
|
pthread_exit(NULL); |
|
return NULL; |
return NULL; |
} |
} |
|
|
static void * | void * |
startSession(sched_task_t *task) |
startSession(sched_task_t *task) |
{ |
{ |
u_char basebuf[USHRT_MAX]; |
u_char basebuf[USHRT_MAX]; |
mqtt_msg_t msg = { NULL, 0 }, buf = { basebuf, sizeof basebuf }; | mqtt_msg_t buf = { basebuf, sizeof basebuf }; |
mqtthdr_connflgs_t flg; |
mqtthdr_connflgs_t flg; |
mqtthdr_connack_t cack; |
mqtthdr_connack_t cack; |
struct tagSession *s, *sess = NULL; |
struct tagSession *s, *sess = NULL; |
int ret; | int ret, wlen; |
|
|
ioTRACE(4); |
ioTRACE(4); |
|
|
assert(task); |
assert(task); |
printf("aaaaaaaaaaaaaaaaa\n"); |
|
fflush(stdout); |
|
|
|
if (!TASK_DATA(task)) { |
if (!TASK_DATA(task)) { |
/* flow from accept new clients */ |
/* flow from accept new clients */ |
sess = initSession(TASK_FD(task), TASK_ARG(task)); |
sess = initSession(TASK_FD(task), TASK_ARG(task)); |
|
io_freeVar(TASK_ARG(task)); |
if (!sess) { |
if (!sess) { |
io_freeVar(TASK_ARG(task)); |
|
close(TASK_FD(task)); |
close(TASK_FD(task)); |
return NULL; |
return NULL; |
} |
} |
Line 310 startSession(sched_task_t *task)
|
Line 200 startSession(sched_task_t *task)
|
} |
} |
} else { |
} else { |
sess = TASK_DATA(task); |
sess = TASK_DATA(task); |
|
assert(sess); |
|
|
buf.msg_len = TASK_DATLEN(task) > sizeof basebuf ? sizeof basebuf : TASK_DATLEN(task); |
buf.msg_len = TASK_DATLEN(task) > sizeof basebuf ? sizeof basebuf : TASK_DATLEN(task); |
memcpy(buf.msg_base, sess->sess_buf->msg_base, buf.msg_len); |
memcpy(buf.msg_base, sess->sess_buf->msg_base, buf.msg_len); |
} |
} |
Line 329 startSession(sched_task_t *task)
|
Line 221 startSession(sched_task_t *task)
|
sess->sess_will.flag = flg.will_flg; |
sess->sess_will.flag = flg.will_flg; |
} |
} |
|
|
printf("sql=%p\n", acc); |
|
fflush(stdout); |
|
/* check online table for user */ |
/* check online table for user */ |
if (call.LoginACC(&cfg, acc, sess->sess_user, sess->sess_pass) < 1) { |
if (call.LoginACC(&cfg, acc, sess->sess_user, sess->sess_pass) < 1) { |
ioDEBUG(0, "Login:: DENIED for username %s and password %s", |
ioDEBUG(0, "Login:: DENIED for username %s and password %s", |
Line 341 startSession(sched_task_t *task)
|
Line 231 startSession(sched_task_t *task)
|
ioDEBUG(1, "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; |
} |
} |
printf(".sql=%p\n", pub); | |
fflush(stdout); | |
if (call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%") > 0) { |
if (call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%") > 0) { |
ioDEBUG(2, "Old session %s should be disconnect!", sess->sess_cid); |
ioDEBUG(2, "Old session %s should be disconnect!", sess->sess_cid); |
TAILQ_FOREACH(s, &Sessions, sess_node) |
TAILQ_FOREACH(s, &Sessions, sess_node) |
if (!strcmp(s->sess_cid, sess->sess_cid)) { |
if (!strcmp(s->sess_cid, sess->sess_cid)) { |
/* found stale session & disconnect it! */ |
/* found stale session & disconnect it! */ |
stopSession(s); | schedWrite(root, leaveClient, sess, TASK_FD(task), NULL, 0); |
break; |
break; |
} |
} |
} |
} |
printf("...sql=%p\n", pub); | |
fflush(stdout); | |
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, |
sess->sess_will.qos, sess->sess_will.retain) == -1) { |
sess->sess_will.qos, sess->sess_will.retain) == -1) { |
Line 366 startSession(sched_task_t *task)
|
Line 254 startSession(sched_task_t *task)
|
ret = MQTT_RETCODE_ACCEPTED; |
ret = MQTT_RETCODE_ACCEPTED; |
} |
} |
|
|
ret = mqtt_msgCONNACK(&msg, ret); | /* Start session task OK ... */ |
if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) { | if (!schedRead(root, dispatchSession, sess, TASK_FD(task), NULL, 0)) { |
ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); | ioLIBERR(sched); |
finiSession(sess); | ret = MQTT_RETCODE_DENIED; |
return NULL; | } else |
} else { | TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); |
ioDEBUG(5, "Sended %d bytes", ret); | |
free(msg.msg_base); | |
memset(&msg, 0, sizeof msg); | |
} | |
|
|
/* Start session thread OK ... */ |
|
SESS_LOCK; |
|
TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); |
|
pthread_create(&sess->sess_tid, NULL, (void*(*)(void*)) thrSession, sess); |
|
pthread_detach(sess->sess_tid); |
|
SESS_UNLOCK; |
|
|
|
call.LOG(logg, "Session %s started from %s for user %s (timeout=%d) OK!\n", sess->sess_cid, |
call.LOG(logg, "Session %s started from %s for user %s (timeout=%d) OK!\n", sess->sess_cid, |
sess->sess_addr, sess->sess_user, sess->sess_ka); |
sess->sess_addr, sess->sess_user, sess->sess_ka); |
return NULL; |
|
end: /* close client connection */ |
end: /* close client connection */ |
ret = mqtt_msgCONNACK(&msg, ret); | wlen = mqtt_msgCONNACK(sess->sess_buf, ret); |
if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) { | if ((wlen = send(TASK_FD(task), sess->sess_buf->msg_base, wlen, 0)) == -1) { |
ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); | ioDEBUG(3, "Error:: send(%d) #%d - %s", TASK_FD(task), errno, strerror(errno)); |
} else { | } else |
ioDEBUG(5, "Sended %d bytes", ret); | ioDEBUG(5, "Sended %d bytes", wlen); |
free(msg.msg_base); | |
memset(&msg, 0, sizeof msg); | |
} | |
|
|
ioDEBUG(1, "Close client %s with socket=%d", sess->sess_addr, sess->sess_sock); | if (ret != MQTT_RETCODE_ACCEPTED) { |
finiSession(sess); | ioDEBUG(1, "Close client %s with socket=%d", sess->sess_addr, TASK_FD(task)); |
| finiSession(sess); |
| } |
return NULL; |
return NULL; |
} |
} |
|
|
Line 418 acceptClient(sched_task_t *task)
|
Line 293 acceptClient(sched_task_t *task)
|
if ((cli = accept(TASK_FD(task), &sa.sa, &sslen)) == -1) |
if ((cli = accept(TASK_FD(task), &sa.sa, &sslen)) == -1) |
goto end; |
goto end; |
else |
else |
fcntl(TASK_FD(task), F_SETFL, fcntl(TASK_FD(task), F_GETFL, 0) | O_NONBLOCK); | fcntl(cli, F_SETFL, fcntl(cli, F_GETFL, 0) | O_NONBLOCK); |
|
|
v = io_allocVar(); |
v = io_allocVar(); |
if (!v) { |
if (!v) { |
Line 428 acceptClient(sched_task_t *task)
|
Line 303 acceptClient(sched_task_t *task)
|
} else { |
} else { |
memset(str, 0, sizeof str); |
memset(str, 0, sizeof str); |
snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa)); |
snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa)); |
|
AIT_FREE_VAL(v); |
AIT_SET_STR(v, str); |
AIT_SET_STR(v, str); |
} |
} |
ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v)); |
ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v)); |
Line 438 acceptClient(sched_task_t *task)
|
Line 314 acceptClient(sched_task_t *task)
|
ioDEBUG(1, "Terminated client with socket=%d", cli); |
ioDEBUG(1, "Terminated client with socket=%d", cli); |
} |
} |
end: |
end: |
if (!schedRead(TASK_ROOT(task), acceptClient, NULL, TASK_FD(task), NULL, 0)) | if (!schedRead(root, acceptClient, NULL, TASK_FD(task), NULL, 0)) |
ioLIBERR(sched); |
ioLIBERR(sched); |
return NULL; |
return NULL; |
} |
} |
Line 449 int
|
Line 325 int
|
Run(int sock) |
Run(int sock) |
{ |
{ |
struct tagPub *pub; |
struct tagPub *pub; |
|
struct tagSession *sess; |
struct timespec pl = { 0, 100000000 }; |
struct timespec pl = { 0, 100000000 }; |
|
|
ioTRACE(1); |
ioTRACE(1); |
Line 465 Run(int sock)
|
Line 342 Run(int sock)
|
return -1; |
return -1; |
} |
} |
|
|
pthread_attr_init(&attr); |
|
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); |
|
|
|
schedPolling(root, &pl, NULL); |
schedPolling(root, &pl, NULL); |
schedRun(root, &Kill); |
schedRun(root, &Kill); |
|
|
pthread_attr_destroy(&attr); |
|
|
|
/* free all undeleted elements into lists */ |
/* free all undeleted elements into lists */ |
PUBS_LOCK; |
|
TAILQ_FOREACH(pub, &Pubs, pub_node) { |
TAILQ_FOREACH(pub, &Pubs, pub_node) { |
TAILQ_REMOVE(&Pubs, pub, pub_node); |
TAILQ_REMOVE(&Pubs, pub, pub_node); |
|
|
Line 482 Run(int sock)
|
Line 353 Run(int sock)
|
if (pub->pub_packet.msg_base) |
if (pub->pub_packet.msg_base) |
free(pub->pub_packet.msg_base); |
free(pub->pub_packet.msg_base); |
} |
} |
PUBS_UNLOCK; | TAILQ_FOREACH(sess, &Sessions, sess_node) { |
| TAILQ_REMOVE(&Sessions, sess, sess_node); |
| |
| finiSession(sess); |
| } |
return 0; |
return 0; |
} |
} |