version 1.2.2.42, 2012/06/29 15:43:13
|
version 1.2.2.47, 2012/07/03 08:39:39
|
Line 70 finiSession(struct tagSession *sess)
|
Line 70 finiSession(struct tagSession *sess)
|
if (!sess) |
if (!sess) |
return; |
return; |
|
|
|
schedCancelby(root, taskTIMER, CRITERIA_CALL, sendRetain, NULL); |
|
|
if (sess->sess_clean) { |
if (sess->sess_clean) { |
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, "%"); |
if (call.DeletePUB_subscribe) |
if (call.DeletePUB_subscribe) |
call.DeletePUB_subscribe(&cfg, pub, sess->sess_cid, "%", sess->sess_user, "%"); |
call.DeletePUB_subscribe(&cfg, pub, sess->sess_cid, "%", sess->sess_user, "%"); |
if (call.WipePUB_topic) | if (call.WipePUB_topic) /* only dynamic messages */ |
call.WipePUB_topic(&cfg, pub, sess->sess_cid, sess->sess_user, -1); | call.WipePUB_topic(&cfg, pub, sess->sess_cid, sess->sess_user, 0); |
} |
} |
|
|
while ((store = SLIST_FIRST(&sess->sess_subscr))) { |
while ((store = SLIST_FIRST(&sess->sess_subscr))) { |
Line 91 finiSession(struct tagSession *sess)
|
Line 93 finiSession(struct tagSession *sess)
|
} |
} |
|
|
if (sess->sess_will.flag) |
if (sess->sess_will.flag) |
srv_Will(sess); | pubWill(sess); |
|
|
if (sess->sess_will.topic) |
if (sess->sess_will.topic) |
free(sess->sess_will.topic); |
free(sess->sess_will.topic); |
Line 219 startSession(sched_task_t *task)
|
Line 221 startSession(sched_task_t *task)
|
struct tagSession *s, *sess = NULL; |
struct tagSession *s, *sess = NULL; |
int ret, wlen; |
int ret, wlen; |
mqtt_subscr_t *subs; |
mqtt_subscr_t *subs; |
|
struct timespec ts = { RETAIN_TIMEOUT, 0 }; |
|
intptr_t sock; |
|
|
ioTRACE(4); |
ioTRACE(4); |
|
|
Line 282 startSession(sched_task_t *task)
|
Line 286 startSession(sched_task_t *task)
|
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! */ |
schedCancelby(root, taskMAX, CRITERIA_FD, (void*) s->sess_sock, NULL); | sock = s->sess_sock; |
| schedCancelby(root, taskMAX, CRITERIA_FD, (void*) sock, NULL); |
schedWrite(root, leaveClient, s, s->sess_sock, NULL, 0); |
schedWrite(root, leaveClient, s, s->sess_sock, NULL, 0); |
break; |
break; |
} |
} |
Line 302 startSession(sched_task_t *task)
|
Line 307 startSession(sched_task_t *task)
|
if (sess->sess_clean) { |
if (sess->sess_clean) { |
if (call.DeletePUB_subscribe) |
if (call.DeletePUB_subscribe) |
call.DeletePUB_subscribe(&cfg, pub, sess->sess_cid, "%", sess->sess_user, "%"); |
call.DeletePUB_subscribe(&cfg, pub, sess->sess_cid, "%", sess->sess_user, "%"); |
if (call.WipePUB_topic) | if (call.WipePUB_topic) /* delete ALL messages */ |
call.WipePUB_topic(&cfg, pub, sess->sess_cid, sess->sess_user, -1); |
call.WipePUB_topic(&cfg, pub, sess->sess_cid, sess->sess_user, -1); |
} else if (call.ReadPUB_subscribe) { |
} else if (call.ReadPUB_subscribe) { |
|
/* load subscribes */ |
subs = call.ReadPUB_subscribe(&cfg, pub, sess->sess_cid, "%"); |
subs = call.ReadPUB_subscribe(&cfg, pub, sess->sess_cid, "%"); |
loadSubscribes(sess, subs); |
loadSubscribes(sess, subs); |
mqtt_subFree(&subs); |
mqtt_subFree(&subs); |
} |
} |
|
|
|
/* timer event for retain messages */ |
|
if (call.ReadPUB_topic) |
|
schedTimer(root, sendRetain, sess, ts, NULL, 0); |
|
|
/* Start session task OK ... */ |
/* Start session task OK ... */ |
if (!schedRead(root, dispatchSession, sess, TASK_FD(task), NULL, 0)) { |
if (!schedRead(root, dispatchSession, sess, TASK_FD(task), NULL, 0)) { |
ioLIBERR(sched); |
ioLIBERR(sched); |
Line 396 Run(int sock)
|
Line 406 Run(int sock)
|
|
|
schedPolling(root, &pl, NULL); |
schedPolling(root, &pl, NULL); |
schedRun(root, &Kill); |
schedRun(root, &Kill); |
|
|
|
schedCancelby(root, taskTIMER, CRITERIA_CALL, sendRetain, NULL); |
|
|
/* free all undeleted elements into lists */ |
/* free all undeleted elements into lists */ |
TAILQ_FOREACH(sess, &Sessions, sess_node) { |
TAILQ_FOREACH(sess, &Sessions, sess_node) { |