version 1.2.2.9, 2012/04/24 13:54:50
|
version 1.2.2.15, 2012/04/25 16:36:26
|
Line 6
|
Line 6
|
|
|
|
|
static void *startSession(sched_task_t *task); |
static void *startSession(sched_task_t *task); |
|
static pthread_attr_t attr; |
|
|
|
|
static inline struct tagSession * |
static inline struct tagSession * |
Line 37 initSession(int sock, ait_val_t * __restrict v)
|
Line 38 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); |
io_freeVar(v); |
return NULL; |
return NULL; |
Line 58 initSession(int sock, ait_val_t * __restrict v)
|
Line 51 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); |
io_freeVar(v); |
return NULL; |
return NULL; |
Line 106 finiSession(struct tagSession *sess)
|
Line 98 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 147 stopSession(struct tagSession *sess)
|
Line 137 stopSession(struct tagSession *sess)
|
sess->sess_addr, sess->sess_user); |
sess->sess_addr, sess->sess_user); |
} |
} |
|
|
static int |
|
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 * |
thrSession(struct tagSession *sess) |
thrSession(struct tagSession *sess) |
{ |
{ |
Line 211 thrSession(struct tagSession *sess)
|
Line 155 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 238 thrSession(struct tagSession *sess)
|
Line 182 thrSession(struct tagSession *sess)
|
io_freeVar(v); |
io_freeVar(v); |
} else |
} else |
ioLIBERR(mqtt); |
ioLIBERR(mqtt); |
|
|
|
SESS_LOCK; |
|
TAILQ_REMOVE(&Sessions, sess, sess_node); |
|
SESS_UNLOCK; |
|
|
locKill ^= locKill; |
locKill ^= locKill; |
continue; | break; |
case MQTT_TYPE_DISCONNECT: |
case MQTT_TYPE_DISCONNECT: |
ioDEBUG(5, "Exec DISCONNECT session"); |
ioDEBUG(5, "Exec DISCONNECT session"); |
|
SESS_LOCK; |
|
TAILQ_REMOVE(&Sessions, sess, sess_node); |
|
SESS_UNLOCK; |
|
|
finiSession(sess); |
finiSession(sess); |
locKill ^= locKill; |
locKill ^= locKill; |
continue; |
continue; |
Line 261 thrSession(struct tagSession *sess)
|
Line 214 thrSession(struct tagSession *sess)
|
case MQTT_TYPE_PINGREQ: |
case MQTT_TYPE_PINGREQ: |
ioDEBUG(5, "Exec PINGREQ session"); |
ioDEBUG(5, "Exec PINGREQ session"); |
break; |
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 270 thrSession(struct tagSession *sess)
|
Line 226 thrSession(struct tagSession *sess)
|
|
|
pthread_cleanup_pop(locKill); |
pthread_cleanup_pop(locKill); |
pthread_exit(NULL); |
pthread_exit(NULL); |
|
return NULL; |
} |
} |
|
|
static void * |
static void * |
Line 281 startSession(sched_task_t *task)
|
Line 238 startSession(sched_task_t *task)
|
mqtthdr_connack_t cack; |
mqtthdr_connack_t cack; |
struct tagSession *s, *sess = NULL; |
struct tagSession *s, *sess = NULL; |
int ret; |
int ret; |
pthread_attr_t attr; |
|
|
|
ioTRACE(4); |
ioTRACE(4); |
|
|
Line 333 startSession(sched_task_t *task)
|
Line 289 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; |
} |
} |
|
|
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 342 startSession(sched_task_t *task)
|
Line 299 startSession(sched_task_t *task)
|
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) { |
Line 366 startSession(sched_task_t *task)
|
Line 324 startSession(sched_task_t *task)
|
} |
} |
|
|
/* Start session thread OK ... */ |
/* Start session thread OK ... */ |
pthread_attr_init(&attr); |
|
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); |
|
|
|
SESS_LOCK; |
SESS_LOCK; |
TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); |
TAILQ_INSERT_TAIL(&Sessions, sess, sess_node); |
pthread_create(&sess->sess_tid, &attr, (void*(*)(void*)) thrSession, sess); |
pthread_create(&sess->sess_tid, &attr, (void*(*)(void*)) thrSession, sess); |
Line 376 startSession(sched_task_t *task)
|
Line 331 startSession(sched_task_t *task)
|
|
|
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); |
|
|
pthread_attr_destroy(&attr); |
|
return NULL; |
return NULL; |
end: /* close client connection */ |
end: /* close client connection */ |
ret = mqtt_msgCONNACK(&msg, ret); |
ret = mqtt_msgCONNACK(&msg, ret); |
Line 430 acceptClient(sched_task_t *task)
|
Line 383 acceptClient(sched_task_t *task)
|
ioDEBUG(1, "Terminated client with socket=%d", cli); |
ioDEBUG(1, "Terminated client with socket=%d", cli); |
} |
} |
end: |
end: |
schedRead(TASK_ROOT(task), acceptClient, NULL, TASK_FD(task), NULL, 0); | if (!schedRead(TASK_ROOT(task), acceptClient, NULL, TASK_FD(task), NULL, 0)) |
| ioLIBERR(sched); |
return NULL; |
return NULL; |
} |
} |
|
|
Line 456 Run(int sock)
|
Line 410 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; |
PUBS_LOCK; |