version 1.2.2.11, 2012/04/25 12:04:30
|
version 1.2.2.21, 2012/05/03 14:00:33
|
Line 23 initSession(int sock, ait_val_t * __restrict v)
|
Line 23 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)); |
Line 38 initSession(int sock, ait_val_t * __restrict v)
|
Line 37 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 49 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 63 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 93 finiSession(struct tagSession *sess)
|
Line 80 finiSession(struct tagSession *sess)
|
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); |
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); |
|
|
SESS_ELEM_LOCK(sess); |
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 107 finiSession(struct tagSession *sess)
|
Line 93 finiSession(struct tagSession *sess)
|
SESS_ELEM_UNLOCK(sess); |
SESS_ELEM_UNLOCK(sess); |
pthread_mutex_destroy(&sess->sess_mtx); |
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 142 stopSession(struct tagSession *sess)
|
Line 126 stopSession(struct tagSession *sess)
|
free(msg.msg_base); |
free(msg.msg_base); |
|
|
ioDEBUG(1, "Close socket=%d", sess->sess_sock); |
ioDEBUG(1, "Close socket=%d", sess->sess_sock); |
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); |
} |
|
|
|
static int | finiSession(sess); |
KASession(struct tagSession *sess) | |
{ | |
mqtt_msg_t msg = { NULL, 0 }; | |
int ret; | |
struct pollfd pfd; | |
| |
ioTRACE(4); | |
| |
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 */ | |
if (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)); | |
return -1; | |
} | |
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! */ | |
return 0; | |
} |
} |
|
|
static void * |
static void * |
Line 200 thrSession(struct tagSession *sess)
|
Line 138 thrSession(struct tagSession *sess)
|
int ret, locKill = 42; |
int ret, locKill = 42; |
struct pollfd pfd; |
struct pollfd pfd; |
struct mqtthdr *hdr; |
struct mqtthdr *hdr; |
ait_val_t *v; |
|
|
|
pthread_cleanup_push((void(*)(void*)) stopSession, sess); |
pthread_cleanup_push((void(*)(void*)) stopSession, sess); |
ioTRACE(2); |
ioTRACE(2); |
Line 212 thrSession(struct tagSession *sess)
|
Line 149 thrSession(struct tagSession *sess)
|
pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { |
pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { |
ioDEBUG(3, "Error:: poll(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); |
ioDEBUG(3, "Error:: poll(%d) #%d - %s", sess->sess_sock, errno, strerror(errno)); |
break; |
break; |
} else if (!ret && (ret = KASession(sess))) { | } else if (!ret && (ret = mqtt_KeepAlive(sess->sess_sock, sess->sess_ka, 1))) { |
call.LOG(logg, "Session %s keep-alive missing from %s for user %s ...\n", |
call.LOG(logg, "Session %s keep-alive missing from %s for user %s ...\n", |
sess->sess_cid, sess->sess_addr, sess->sess_user); |
sess->sess_cid, sess->sess_addr, sess->sess_user); |
break; |
break; |
Line 230 thrSession(struct tagSession *sess)
|
Line 167 thrSession(struct tagSession *sess)
|
/* dispatch message type */ |
/* dispatch message type */ |
if (mqtt_srv_Dispatch(sess->sess_srv, sess)) |
if (mqtt_srv_Dispatch(sess->sess_srv, sess)) |
ioLIBERR(mqtt); |
ioLIBERR(mqtt); |
locKill ^= locKill; |
|
switch (hdr->mqtt_msg.type) { |
switch (hdr->mqtt_msg.type) { |
case MQTT_TYPE_CONNECT: |
case MQTT_TYPE_CONNECT: |
ioDEBUG(5, "Exec CONNECT session"); | schedEvent(root, startSession, NULL, (u_long) sess->sess_sock, sess, ret); |
if ((v = io_allocVar())) { | locKill ^= locKill; |
AIT_SET_STR(v, sess->sess_addr); | break; |
if (!schedEvent(root, startSession, v, (u_long) sess->sess_sock, sess, ret)) | |
io_freeVar(v); | |
} else | |
ioLIBERR(mqtt); | |
continue; | |
case MQTT_TYPE_DISCONNECT: |
case MQTT_TYPE_DISCONNECT: |
ioDEBUG(5, "Exec DISCONNECT session"); |
|
finiSession(sess); |
finiSession(sess); |
locKill = 42; | locKill ^= locKill; |
continue; |
continue; |
|
case MQTT_TYPE_SUBSCRIBE: |
|
case MQTT_TYPE_PINGREQ: |
|
break; |
case MQTT_TYPE_PUBLISH: |
case MQTT_TYPE_PUBLISH: |
ioDEBUG(5, "Exec PUBLISH topic QoS=%d", hdr->mqtt_msg.qos); |
ioDEBUG(5, "Exec PUBLISH topic QoS=%d", hdr->mqtt_msg.qos); |
/* |
/* |
Line 255 thrSession(struct tagSession *sess)
|
Line 188 thrSession(struct tagSession *sess)
|
break; |
break; |
case MQTT_TYPE_PUBREL: |
case MQTT_TYPE_PUBREL: |
break; |
break; |
case MQTT_TYPE_SUBSCRIBE: |
|
break; |
|
case MQTT_TYPE_UNSUBSCRIBE: |
case MQTT_TYPE_UNSUBSCRIBE: |
break; |
break; |
case MQTT_TYPE_PINGREQ: |
|
ioDEBUG(5, "Exec PINGREQ session"); |
|
break; |
|
case MQTT_TYPE_PINGRESP: |
|
ioDEBUG(5, "Exec PINGRESP session"); |
|
break; |
|
default: |
default: |
ioDEBUG(5, "Error:: Session %s, wrong command %d - DISCARDED", |
ioDEBUG(5, "Error:: Session %s, wrong command %d - DISCARDED", |
sess->sess_cid, hdr->mqtt_msg.type); |
sess->sess_cid, hdr->mqtt_msg.type); |
Line 290 startSession(sched_task_t *task)
|
Line 215 startSession(sched_task_t *task)
|
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 329 startSession(sched_task_t *task)
|
Line 252 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 262 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) |
Line 352 startSession(sched_task_t *task)
|
Line 272 startSession(sched_task_t *task)
|
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 380 startSession(sched_task_t *task)
|
Line 299 startSession(sched_task_t *task)
|
/* Start session thread OK ... */ |
/* Start session thread OK ... */ |
SESS_LOCK; |
SESS_LOCK; |
TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); |
TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); |
pthread_create(&sess->sess_tid, NULL, (void*(*)(void*)) thrSession, sess); | pthread_create(&sess->sess_tid, &attr, (void*(*)(void*)) thrSession, sess); |
pthread_detach(sess->sess_tid); | |
SESS_UNLOCK; |
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, |
Line 418 acceptClient(sched_task_t *task)
|
Line 336 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 346 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)); |