version 1.2.2.10, 2012/05/05 14:51:02
|
version 1.2.2.16, 2012/05/08 14:36:10
|
Line 72 cmdSUBSCRIBE(void *srv, int len, void *arg)
|
Line 72 cmdSUBSCRIBE(void *srv, int len, void *arg)
|
|
|
/* add to db */ |
/* add to db */ |
for (i = 0; i < siz; i++) { |
for (i = 0; i < siz; i++) { |
if (call.WritePUB_subscribe(&cfg, pub, mid, subs[i].sub_topic.msg_base, | if (call.WritePUB_subscribe(&cfg, pub, sess->sess_cid, mid, subs[i].sub_topic.msg_base, |
sess->sess_user, sess->sess_addr, subs[i].sub_ret) > 0) { |
sess->sess_user, sess->sess_addr, subs[i].sub_ret) > 0) { |
store = malloc(sizeof(struct tagStore)); |
store = malloc(sizeof(struct tagStore)); |
if (!store) { |
if (!store) { |
Line 97 cmdSUBSCRIBE(void *srv, int len, void *arg)
|
Line 97 cmdSUBSCRIBE(void *srv, int len, void *arg)
|
ioDEBUG(5, "Error:: in msgSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
ioDEBUG(5, "Error:: in msgSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
goto end; |
goto end; |
} |
} |
if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, 0)) == -1) | if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, MSG_NOSIGNAL)) == -1) |
ioSYSERR(0); |
ioSYSERR(0); |
else { |
else { |
ioDEBUG(5, "Sended %d bytes.", siz); |
ioDEBUG(5, "Sended %d bytes.", siz); |
Line 112 int
|
Line 112 int
|
cmdUNSUBSCRIBE(void *srv, int len, void *arg) |
cmdUNSUBSCRIBE(void *srv, int len, void *arg) |
{ |
{ |
struct tagSession *sess = (struct tagSession*) arg; |
struct tagSession *sess = (struct tagSession*) arg; |
|
mqtt_subscr_t *subs = NULL; |
|
int siz = 0; |
|
u_short mid = 0; |
|
register int i; |
|
struct tagStore *store, *tmp; |
|
|
ioTRACE(2); |
ioTRACE(2); |
|
|
if (!sess) |
if (!sess) |
return -1; |
return -1; |
|
|
|
ioDEBUG(5, "Exec UNSUBSCRIBE session"); |
|
siz = mqtt_readUNSUBSCRIBE(sess->sess_buf, &mid, &subs); |
|
if (siz == -1) { |
|
ioDEBUG(5, "Error:: in readUNSUBSCRIBE #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
|
return 0; |
|
} |
|
|
|
/* del from db */ |
|
for (i = 0; i < siz; i++) { |
|
SLIST_FOREACH_SAFE(store, &sess->sess_subscr, st_node, tmp) { |
|
if (store->st_subscr.sub_ret == subs[i].sub_ret && |
|
store->st_subscr.sub_topic.msg_base && |
|
!strcmp(store->st_subscr.sub_topic.msg_base, |
|
subs[i].sub_topic.msg_base)) { |
|
SLIST_REMOVE(&sess->sess_subscr, store, tagStore, st_node); |
|
|
|
if (store->st_subscr.sub_topic.msg_base) |
|
free(store->st_subscr.sub_topic.msg_base); |
|
if (store->st_subscr.sub_value.msg_base) |
|
free(store->st_subscr.sub_value.msg_base); |
|
free(store); |
|
} |
|
} |
|
|
|
call.DeletePUB_subscribe(&cfg, pub, sess->sess_cid, subs[i].sub_topic.msg_base, |
|
sess->sess_user, "%"); |
|
} |
|
|
|
/* send acknowledge */ |
|
siz = mqtt_msgUNSUBACK(sess->sess_buf, mid); |
|
if (siz == -1) { |
|
ioDEBUG(5, "Error:: in msgUNSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
|
goto end; |
|
} |
|
if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, MSG_NOSIGNAL)) == -1) |
|
ioSYSERR(0); |
|
else { |
|
ioDEBUG(5, "Sended %d bytes.", siz); |
|
memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); |
|
} |
|
end: |
|
mqtt_subFree(&subs); |
return 0; |
return 0; |
} |
} |
|
|
Line 138 cmdPINGREQ(void *srv, int len, void *arg)
|
Line 185 cmdPINGREQ(void *srv, int len, void *arg)
|
ioDEBUG(5, "Error:: in msgPINGRESP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
ioDEBUG(5, "Error:: in msgPINGRESP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); |
return 0; |
return 0; |
} |
} |
if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, 0)) == -1) { | if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, MSG_NOSIGNAL)) == -1) { |
ioSYSERR(0); |
ioSYSERR(0); |
return 0; |
return 0; |
} else { |
} else { |
Line 163 cmdCONNECT(void *srv, int len, void *arg)
|
Line 210 cmdCONNECT(void *srv, int len, void *arg)
|
ioDEBUG(5, "Exec CONNECT session"); |
ioDEBUG(5, "Exec CONNECT session"); |
TAILQ_REMOVE(&Sessions, sess, sess_node); |
TAILQ_REMOVE(&Sessions, sess, sess_node); |
|
|
if (call.FiniSessPUB) | if (sess->sess_clean) { |
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); | 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) |
| call.WipePUB_topic(&cfg, pub, sess->sess_cid, sess->sess_user, -1); |
| } |
|
|
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 185 cmdCONNECT(void *srv, int len, void *arg)
|
Line 238 cmdCONNECT(void *srv, int len, void *arg)
|
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); |
|
|
// schedEvent(root, startSession, NULL, (u_long) sess->sess_sock, sess, ret); | return -3; /* reconnect client */ |
return 0; | |
} |
} |
|
|
int |
int |
Line 200 cmdDISCONNECT(void *srv, int len, void *arg)
|
Line 252 cmdDISCONNECT(void *srv, int len, void *arg)
|
return -1; |
return -1; |
|
|
ioDEBUG(5, "Exec DISCONNECT session"); |
ioDEBUG(5, "Exec DISCONNECT session"); |
|
|
TAILQ_REMOVE(&Sessions, sess, sess_node); |
|
|
|
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); |