version 1.2.2.25, 2012/05/05 14:51:02
|
version 1.5.2.1, 2013/01/18 10:17:22
|
Line 1
|
Line 1
|
|
/************************************************************************* |
|
* (C) 2011 AITNET ltd - Sofia/Bulgaria - <misho@aitbg.com> |
|
* by Michael Pounov <misho@openbsd-bg.org> |
|
* |
|
* $Author$ |
|
* $Id$ |
|
* |
|
************************************************************************** |
|
The ELWIX and AITNET software is distributed under the following |
|
terms: |
|
|
|
All of the documentation and software included in the ELWIX and AITNET |
|
Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org> |
|
|
|
Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013 |
|
by Michael Pounov <misho@elwix.org>. All rights reserved. |
|
|
|
Redistribution and use in source and binary forms, with or without |
|
modification, are permitted provided that the following conditions |
|
are met: |
|
1. Redistributions of source code must retain the above copyright |
|
notice, this list of conditions and the following disclaimer. |
|
2. Redistributions in binary form must reproduce the above copyright |
|
notice, this list of conditions and the following disclaimer in the |
|
documentation and/or other materials provided with the distribution. |
|
3. All advertising materials mentioning features or use of this software |
|
must display the following acknowledgement: |
|
This product includes software developed by Michael Pounov <misho@elwix.org> |
|
ELWIX - Embedded LightWeight unIX and its contributors. |
|
4. Neither the name of AITNET nor the names of its contributors |
|
may be used to endorse or promote products derived from this software |
|
without specific prior written permission. |
|
|
|
THIS SOFTWARE IS PROVIDED BY AITNET AND CONTRIBUTORS ``AS IS'' AND |
|
ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE |
|
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE |
|
ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE |
|
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL |
|
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS |
|
OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) |
|
HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT |
|
LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY |
|
OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF |
|
SUCH DAMAGE. |
|
*/ |
#include "global.h" |
#include "global.h" |
#include "rtlm.h" |
#include "rtlm.h" |
#include "utils.h" |
|
#include "mqttd.h" |
#include "mqttd.h" |
#include "mqttd_calls.h" |
#include "mqttd_calls.h" |
|
#include "utils.h" |
|
|
|
|
static inline struct tagSession * |
static inline struct tagSession * |
Line 11 initSession(int sock, ait_val_t * __restrict v)
|
Line 56 initSession(int sock, ait_val_t * __restrict v)
|
struct tagSession *sess = NULL; |
struct tagSession *sess = NULL; |
const char *str; |
const char *str; |
|
|
ioTRACE(5); | ETRACE(5); |
|
|
if (!v) |
if (!v) |
return NULL; |
return NULL; |
|
|
sess = malloc(sizeof(struct tagSession)); | sess = e_malloc(sizeof(struct tagSession)); |
if (!sess) { |
if (!sess) { |
ioSYSERR(0); | ELIBERR(elwix); |
return NULL; |
return NULL; |
} else |
} else |
memset(sess, 0, sizeof(struct tagSession)); |
memset(sess, 0, sizeof(struct tagSession)); |
Line 33 initSession(int sock, ait_val_t * __restrict v)
|
Line 78 initSession(int sock, ait_val_t * __restrict v)
|
|
|
sess->sess_buf = mqtt_msgAlloc(USHRT_MAX); |
sess->sess_buf = mqtt_msgAlloc(USHRT_MAX); |
if (!sess->sess_buf) { |
if (!sess->sess_buf) { |
ioLIBERR(mqtt); | ELIBERR(mqtt); |
free(sess); | e_free(sess); |
return NULL; |
return NULL; |
} |
} |
|
|
/* init server actor */ |
/* init server actor */ |
sess->sess_srv = mqtt_srv_Init(sock, sess->sess_buf); | sess->sess_srv = mqtt_srv_cliInit(sock, sess->sess_buf, sess->sess_ka, 1); |
if (!sess->sess_srv) { |
if (!sess->sess_srv) { |
ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError()); | EVERBOSE(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
mqtt_msgFree(&sess->sess_buf, 42); |
mqtt_msgFree(&sess->sess_buf, 42); |
free(sess); | e_free(sess); |
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 65 finiSession(struct tagSession *sess)
|
Line 110 finiSession(struct tagSession *sess)
|
{ |
{ |
struct tagStore *store; |
struct tagStore *store; |
|
|
ioTRACE(5); | ETRACE(5); |
|
|
if (!sess) |
if (!sess) |
return; |
return; |
|
|
if (call.FiniSessPUB) | schedCancelby(root, taskTIMER, CRITERIA_CALL, sendRetain, NULL); |
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); | |
|
|
|
if (sess->sess_clean) { |
|
if (call.FiniSessPUB) |
|
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); |
|
if (call.DeletePUB_subscribe) |
|
call.DeletePUB_subscribe(&cfg, pub, sess->sess_cid, "%", sess->sess_user, "%"); |
|
if (call.WipePUB_topic) /* only dynamic messages */ |
|
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))) { |
SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node); |
SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node); |
|
|
Line 81 finiSession(struct tagSession *sess)
|
Line 134 finiSession(struct tagSession *sess)
|
if (store->st_subscr.sub_value.msg_base) |
if (store->st_subscr.sub_value.msg_base) |
free(store->st_subscr.sub_value.msg_base); |
free(store->st_subscr.sub_value.msg_base); |
|
|
free(store); | e_free(store); |
} |
} |
|
|
if (sess->sess_will.msg) | if (sess->sess_will.flag) |
free(sess->sess_will.msg); | pubWill(sess); |
| |
if (sess->sess_will.topic) |
if (sess->sess_will.topic) |
free(sess->sess_will.topic); |
free(sess->sess_will.topic); |
|
if (sess->sess_will.msg) |
|
free(sess->sess_will.msg); |
|
|
if (sess->sess_sock > STDERR_FILENO) |
if (sess->sess_sock > STDERR_FILENO) |
srv_Close(sess->sess_sock); |
srv_Close(sess->sess_sock); |
|
|
mqtt_srv_Fini(&sess->sess_srv); | mqtt_srv_cliFini(&sess->sess_srv); |
mqtt_msgFree(&sess->sess_buf, 42); |
mqtt_msgFree(&sess->sess_buf, 42); |
|
|
free(sess); | e_free(sess); |
} |
} |
|
|
static void * |
static void * |
Line 104 leaveClient(sched_task_t *task)
|
Line 160 leaveClient(sched_task_t *task)
|
struct tagSession *sess; |
struct tagSession *sess; |
int ret; |
int ret; |
|
|
ioTRACE(4); | ETRACE(4); |
|
|
assert(task); |
assert(task); |
|
|
Line 116 leaveClient(sched_task_t *task)
|
Line 172 leaveClient(sched_task_t *task)
|
ret = mqtt_msgDISCONNECT(sess->sess_buf); |
ret = mqtt_msgDISCONNECT(sess->sess_buf); |
send(TASK_FD(task), sess->sess_buf->msg_base, ret, MSG_NOSIGNAL); |
send(TASK_FD(task), sess->sess_buf->msg_base, ret, MSG_NOSIGNAL); |
|
|
ioDEBUG(1, "Close socket=%d", TASK_FD(task)); | EVERBOSE(1, "Close socket=%ld", (long) TASK_FD(task)); |
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); |
|
|
Line 130 dispatchSession(sched_task_t *task)
|
Line 186 dispatchSession(sched_task_t *task)
|
int ret, len = 0; |
int ret, len = 0; |
struct tagSession *sess; |
struct tagSession *sess; |
|
|
ioTRACE(2); | ETRACE(2); |
|
|
assert(task); |
assert(task); |
|
|
Line 139 dispatchSession(sched_task_t *task)
|
Line 195 dispatchSession(sched_task_t *task)
|
|
|
/* receive & decode packet */ |
/* receive & decode packet */ |
if ((ret = recv(TASK_FD(task), 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)); | EVERBOSE(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); |
| TAILQ_REMOVE(&Sessions, sess, sess_node); |
finiSession(sess); |
finiSession(sess); |
return NULL; |
return NULL; |
} else if (!ret) { |
} else if (!ret) { |
ioDEBUG(4, "Session %s EOF received.", sess->sess_cid); | EVERBOSE(4, "Session %s EOF received.", sess->sess_cid); |
| TAILQ_REMOVE(&Sessions, sess, sess_node); |
finiSession(sess); |
finiSession(sess); |
return NULL; |
return NULL; |
} |
} |
|
|
do { |
do { |
/* dispatch message type */ |
/* dispatch message type */ |
if ((len = mqtt_srv_Dispatch(sess->sess_srv, ret, sess)) < 0) { | if ((len = mqtt_srv_cliDispatch(sess->sess_srv, ret, sess)) < 0) { |
if (len == -1) | if (len == -1) { |
ioLIBERR(mqtt); | ELIBERR(mqtt); |
finiSession(sess); | finiSession(sess); |
| } else if (len == -2) { |
| TAILQ_REMOVE(&Sessions, sess, sess_node); |
| finiSession(sess); |
| } else if (len == -3) |
| schedEvent(root, startSession, NULL, (u_long) TASK_FD(task), sess, ret); |
} else |
} else |
ret -= len; |
ret -= len; |
} while (len > 0 && ret > 0); |
} while (len > 0 && ret > 0); |
|
|
if (len >= 0 && !schedRead(root, dispatchSession, TASK_ARG(task), TASK_FD(task), NULL, 0)) { |
if (len >= 0 && !schedRead(root, dispatchSession, TASK_ARG(task), TASK_FD(task), NULL, 0)) { |
ioLIBERR(sched); | ELIBERR(sched); |
| TAILQ_REMOVE(&Sessions, sess, sess_node); |
finiSession(sess); |
finiSession(sess); |
} |
} |
return NULL; |
return NULL; |
} |
} |
|
|
|
static int |
|
loadSubscribes(struct tagSession * __restrict sess, mqtt_subscr_t * __restrict subs) |
|
{ |
|
register int i; |
|
struct tagStore *store; |
|
|
|
if (!subs) |
|
return -1; |
|
|
|
for (i = 0; subs[i].sub_topic.msg_base; i++) { |
|
store = e_malloc(sizeof(struct tagStore)); |
|
if (!store) { |
|
ELIBERR(elwix); |
|
continue; |
|
} else { |
|
store->st_msgid = 0; |
|
mqtt_subCopy(&store->st_subscr, &subs[i]); |
|
} |
|
|
|
/* add to cache */ |
|
SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node); |
|
} |
|
|
|
return 0; |
|
} |
|
|
void * |
void * |
startSession(sched_task_t *task) |
startSession(sched_task_t *task) |
{ |
{ |
Line 172 startSession(sched_task_t *task)
|
Line 262 startSession(sched_task_t *task)
|
mqtt_msg_t 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; |
|
ait_val_t *v; |
struct tagSession *s, *sess = NULL; |
struct tagSession *s, *sess = NULL; |
int ret, wlen; |
int ret, wlen; |
|
mqtt_subscr_t *subs; |
|
struct timespec ts = { RETAIN_TIMEOUT, 0 }; |
|
intptr_t sock; |
|
|
ioTRACE(4); | ETRACE(4); |
|
|
assert(task); |
assert(task); |
|
|
if (!TASK_DATA(task)) { |
if (!TASK_DATA(task)) { |
|
v = TASK_ARG(task); |
/* flow from accept new clients */ |
/* flow from accept new clients */ |
sess = initSession(TASK_FD(task), TASK_ARG(task)); | sess = initSession(TASK_FD(task), v); |
io_freeVar(TASK_ARG(task)); | ait_freeVar(&v); |
if (!sess) { |
if (!sess) { |
close(TASK_FD(task)); |
close(TASK_FD(task)); |
return NULL; |
return NULL; |
Line 190 startSession(sched_task_t *task)
|
Line 285 startSession(sched_task_t *task)
|
|
|
/* receive & decode packet */ |
/* receive & decode packet */ |
if (recv(sess->sess_sock, buf.msg_base, buf.msg_len, 0) == -1) { |
if (recv(sess->sess_sock, buf.msg_base, buf.msg_len, 0) == -1) { |
ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); | EVERBOSE(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); |
finiSession(sess); |
finiSession(sess); |
return NULL; |
return NULL; |
} |
} |
Line 208 startSession(sched_task_t *task)
|
Line 303 startSession(sched_task_t *task)
|
ret = cack.retcode; |
ret = cack.retcode; |
flg.flags = cack.reserved; |
flg.flags = cack.reserved; |
if (flg.reserved) { |
if (flg.reserved) { |
ioDEBUG(3, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError()); | EVERBOSE(3, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
goto end; |
goto end; |
} else { |
} else { |
sess->sess_clean = flg.clean_sess; |
sess->sess_clean = flg.clean_sess; |
sess->sess_will.qos = flg.will_qos; |
sess->sess_will.qos = flg.will_qos; |
sess->sess_will.retain = flg.will_retain; |
sess->sess_will.retain = flg.will_retain; |
sess->sess_will.flag = flg.will_flg; |
sess->sess_will.flag = flg.will_flg; |
|
|
|
sess->sess_srv->timeout = sess->sess_ka; |
} |
} |
|
|
/* 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", | EVERBOSE(0, "Login:: DENIED for username %s and password %s", |
sess->sess_user, sess->sess_pass); |
sess->sess_user, sess->sess_pass); |
ret = MQTT_RETCODE_DENIED; |
ret = MQTT_RETCODE_DENIED; |
goto end; |
goto end; |
} else { |
} else { |
ioDEBUG(1, "Login:: ALLOWED for username %s ...", sess->sess_user); | EVERBOSE(1, "Login:: ALLOWED for username %s ...", sess->sess_user); |
ret = MQTT_RETCODE_ACCEPTED; |
ret = MQTT_RETCODE_ACCEPTED; |
} |
} |
|
|
|
/* db management */ |
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); | EVERBOSE(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! */ |
schedWrite(root, leaveClient, sess, TASK_FD(task), NULL, 0); | sock = s->sess_sock; |
| schedCancelby(root, taskMAX, CRITERIA_FD, (void*) sock, NULL); |
| schedWrite(root, leaveClient, s, s->sess_sock, NULL, 0); |
break; |
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, |
sess->sess_will.qos, sess->sess_will.retain) == -1) { |
sess->sess_will.qos, sess->sess_will.retain) == -1) { |
ioDEBUG(0, "Session %s DENIED for username %s", sess->sess_cid, sess->sess_user); | EVERBOSE(0, "Session %s DENIED for username %s", sess->sess_cid, sess->sess_user); |
ret = MQTT_RETCODE_DENIED; |
ret = MQTT_RETCODE_DENIED; |
goto end; |
goto end; |
} else { |
} else { |
ioDEBUG(0, "Session %s from %s and username %s is started", | EVERBOSE(0, "Session %s from %s and username %s is started", |
sess->sess_cid, sess->sess_addr, sess->sess_user); |
sess->sess_cid, sess->sess_addr, sess->sess_user); |
ret = MQTT_RETCODE_ACCEPTED; |
ret = MQTT_RETCODE_ACCEPTED; |
} |
} |
|
/* clean/load session if requested */ |
|
if (sess->sess_clean) { |
|
if (call.DeletePUB_subscribe) |
|
call.DeletePUB_subscribe(&cfg, pub, sess->sess_cid, "%", sess->sess_user, "%"); |
|
if (call.WipePUB_topic) /* delete ALL messages */ |
|
call.WipePUB_topic(&cfg, pub, sess->sess_cid, sess->sess_user, -1); |
|
} else if (call.ReadPUB_subscribe) { |
|
/* load subscribes */ |
|
subs = call.ReadPUB_subscribe(&cfg, pub, sess->sess_cid, "%"); |
|
loadSubscribes(sess, subs); |
|
mqtt_subFree(&subs); |
|
} |
|
|
/* Start session task OK ... */ | /* timer event for retain messages */ |
TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); | if (call.ReadPUB_topic) |
| schedTimer(root, sendRetain, sess, ts, NULL, 0); |
|
|
|
/* 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); | ELIBERR(sched); |
ret = MQTT_RETCODE_DENIED; |
ret = MQTT_RETCODE_DENIED; |
} | } else |
| TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); |
|
|
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); |
end: /* close client connection */ |
end: /* close client connection */ |
wlen = mqtt_msgCONNACK(sess->sess_buf, ret); |
wlen = mqtt_msgCONNACK(sess->sess_buf, ret); |
if ((wlen = send(TASK_FD(task), sess->sess_buf->msg_base, wlen, 0)) == -1) { |
if ((wlen = send(TASK_FD(task), sess->sess_buf->msg_base, wlen, 0)) == -1) { |
ioDEBUG(3, "Error:: send(%d) #%d - %s", TASK_FD(task), errno, strerror(errno)); | EVERBOSE(3, "Error:: send(%ld) #%d - %s", (long) TASK_FD(task), errno, strerror(errno)); |
} else |
} else |
ioDEBUG(5, "Sended %d bytes", wlen); | EVERBOSE(5, "Sended %d bytes", wlen); |
|
|
if (ret != MQTT_RETCODE_ACCEPTED) { |
if (ret != MQTT_RETCODE_ACCEPTED) { |
ioDEBUG(1, "Close client %s with socket=%d", sess->sess_addr, TASK_FD(task)); | EVERBOSE(1, "Close client %s with socket=%ld", sess->sess_addr, (long) TASK_FD(task)); |
finiSession(sess); |
finiSession(sess); |
} |
} |
return NULL; |
return NULL; |
Line 278 static void *
|
Line 392 static void *
|
acceptClient(sched_task_t *task) |
acceptClient(sched_task_t *task) |
{ |
{ |
int cli; |
int cli; |
io_sockaddr_t sa; | sockaddr_t sa; |
socklen_t sslen = sizeof sa.ss; |
socklen_t sslen = sizeof sa.ss; |
ait_val_t *v; |
ait_val_t *v; |
char str[STRSIZ]; |
char str[STRSIZ]; |
|
|
ioTRACE(4); | ETRACE(4); |
|
|
assert(task); |
assert(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(cli, F_SETFL, fcntl(cli, F_GETFL, 0) | O_NONBLOCK); | fcntl(cli, F_SETFL, fcntl(cli, F_GETFL) | O_NONBLOCK); |
|
|
v = io_allocVar(); | v = ait_allocVar(); |
if (!v) { |
if (!v) { |
ioLIBERR(io); | ELIBERR(elwix); |
close(cli); |
close(cli); |
goto end; |
goto end; |
} 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", e_n2addr(&sa, v), e_n2port(&sa)); |
AIT_FREE_VAL(v); |
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)); | EVERBOSE(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v)); |
|
|
if (!schedRead(root, startSession, v, cli, NULL, 0)) { |
if (!schedRead(root, startSession, v, cli, NULL, 0)) { |
io_freeVar(v); | ait_freeVar(&v); |
close(cli); |
close(cli); |
ioDEBUG(1, "Terminated client with socket=%d", cli); | EVERBOSE(1, "Terminated client with socket=%d", cli); |
} |
} |
end: |
end: |
if (!schedRead(root, acceptClient, NULL, TASK_FD(task), NULL, 0)) |
if (!schedRead(root, acceptClient, NULL, TASK_FD(task), NULL, 0)) |
ioLIBERR(sched); | ELIBERR(sched); |
return NULL; |
return NULL; |
} |
} |
|
|
Line 321 end:
|
Line 435 end:
|
int |
int |
Run(int sock) |
Run(int sock) |
{ |
{ |
struct tagPub *pub; |
|
struct tagSession *sess; |
struct tagSession *sess; |
struct timespec pl = { 0, 100000000 }; |
struct timespec pl = { 0, 100000000 }; |
|
|
ioTRACE(1); | ETRACE(1); |
|
|
if (listen(sock, SOMAXCONN) == -1) { | if (mqtt_srv_Listen(sock, 0, 1) == -1) { |
ioSYSERR(0); | ELIBERR(mqtt); |
return -1; |
return -1; |
} else | } |
fcntl(sock, F_SETFL, fcntl(sock, F_GETFL, 0) | O_NONBLOCK); | |
|
|
/* state machine - accept new connections */ |
/* state machine - accept new connections */ |
if (!schedRead(root, acceptClient, NULL, sock, NULL, 0)) { |
if (!schedRead(root, acceptClient, NULL, sock, NULL, 0)) { |
ioLIBERR(sched); | ELIBERR(sched); |
return -1; |
return -1; |
} |
} |
|
|
schedPolling(root, &pl, NULL); |
schedPolling(root, &pl, NULL); |
schedRun(root, &Kill); |
schedRun(root, &Kill); |
|
|
/* free all undeleted elements into lists */ | schedCancelby(root, taskTIMER, CRITERIA_CALL, sendRetain, NULL); |
TAILQ_FOREACH(pub, &Pubs, pub_node) { | |
TAILQ_REMOVE(&Pubs, pub, pub_node); | |
|
|
AIT_FREE_VAL(&pub->pub_name); | /* free all undeleted elements into lists */ |
if (pub->pub_packet.msg_base) | |
free(pub->pub_packet.msg_base); | |
} | |
TAILQ_FOREACH(sess, &Sessions, sess_node) { |
TAILQ_FOREACH(sess, &Sessions, sess_node) { |
TAILQ_REMOVE(&Sessions, sess, sess_node); |
TAILQ_REMOVE(&Sessions, sess, sess_node); |
|
|
finiSession(sess); |
finiSession(sess); |
} |
} |
return 0; |
return 0; |