version 1.2.2.7, 2012/04/27 10:07:51
|
version 1.2.2.11, 2012/05/05 15:04:29
|
Line 5
|
Line 5
|
|
|
|
|
int |
int |
cmdPUBLISH(void *srv, void *arg) | cmdPUBLISH(void *srv, int len, void *arg) |
{ |
{ |
struct mqtthdr *hdr; |
struct mqtthdr *hdr; |
struct tagSession *sess = (struct tagSession*) arg; |
struct tagSession *sess = (struct tagSession*) arg; |
Line 33 cmdPUBLISH(void *srv, void *arg)
|
Line 33 cmdPUBLISH(void *srv, void *arg)
|
} |
} |
|
|
int |
int |
cmdPUBREL(void *srv, void *arg) | cmdPUBREL(void *srv, int len, void *arg) |
{ |
{ |
struct mqtthdr *hdr; |
struct mqtthdr *hdr; |
struct tagSession *sess = (struct tagSession*) arg; |
struct tagSession *sess = (struct tagSession*) arg; |
Line 49 cmdPUBREL(void *srv, void *arg)
|
Line 49 cmdPUBREL(void *srv, void *arg)
|
} |
} |
|
|
int |
int |
cmdSUBSCRIBE(void *srv, void *arg) | cmdSUBSCRIBE(void *srv, int len, void *arg) |
{ |
{ |
struct tagSession *sess = (struct tagSession*) arg; |
struct tagSession *sess = (struct tagSession*) arg; |
mqtt_subscr_t *subs = NULL; |
mqtt_subscr_t *subs = NULL; |
Line 72 cmdSUBSCRIBE(void *srv, void *arg)
|
Line 72 cmdSUBSCRIBE(void *srv, void *arg)
|
|
|
/* add to db */ |
/* add to db */ |
for (i = 0; i < siz; i++) { |
for (i = 0; i < siz; i++) { |
if ((siz = call.WritePUB_subscribe(&cfg, pub, mid, subs[i].sub_topic.msg_base, | if (call.WritePUB_subscribe(&cfg, pub, 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) { |
ioSYSERR(0); |
ioSYSERR(0); |
goto end; |
goto end; |
} else { |
} else { |
store->st_msgid = mid; |
store->st_msgid = mid; |
store->st_subscr = subs[i]; | mqtt_subCopy(&store->st_subscr, &subs[i]); |
} |
} |
|
|
/* add to cache */ |
/* add to cache */ |
SESS_ELEM_LOCK(sess); |
|
SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node); |
SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node); |
SESS_ELEM_UNLOCK(sess); |
|
|
|
subs[i].sub_ret = MQTT_QOS_PASS; |
subs[i].sub_ret = MQTT_QOS_PASS; |
} else |
} else |
Line 101 cmdSUBSCRIBE(void *srv, void *arg)
|
Line 99 cmdSUBSCRIBE(void *srv, void *arg)
|
} |
} |
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, 0)) == -1) |
ioSYSERR(0); |
ioSYSERR(0); |
else | else { |
ioDEBUG(5, "Sended %d bytes.", siz); |
ioDEBUG(5, "Sended %d bytes.", siz); |
|
memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); |
|
} |
end: |
end: |
mqtt_subFree(&subs); |
mqtt_subFree(&subs); |
return 0; |
return 0; |
} |
} |
|
|
int |
int |
cmdUNSUBSCRIBE(void *srv, void *arg) | cmdUNSUBSCRIBE(void *srv, int len, void *arg) |
{ |
{ |
struct tagSession *sess = (struct tagSession*) arg; |
struct tagSession *sess = (struct tagSession*) arg; |
|
|
Line 122 cmdUNSUBSCRIBE(void *srv, void *arg)
|
Line 122 cmdUNSUBSCRIBE(void *srv, void *arg)
|
} |
} |
|
|
int |
int |
cmdPINGREQ(void *srv, void *arg) | cmdPINGREQ(void *srv, int len, void *arg) |
{ |
{ |
struct tagSession *sess = (struct tagSession*) arg; |
struct tagSession *sess = (struct tagSession*) arg; |
int siz = 0; |
int siz = 0; |
Line 141 cmdPINGREQ(void *srv, void *arg)
|
Line 141 cmdPINGREQ(void *srv, void *arg)
|
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, 0)) == -1) { |
ioSYSERR(0); |
ioSYSERR(0); |
return 0; |
return 0; |
} else | } else { |
ioDEBUG(5, "Sended %d bytes.", siz); |
ioDEBUG(5, "Sended %d bytes.", siz); |
|
memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); |
|
} |
|
|
return 0; |
return 0; |
} |
} |
|
|
int |
int |
cmdCONNECT(void *srv, void *arg) | cmdCONNECT(void *srv, int len, void *arg) |
{ |
{ |
struct tagStore *store; |
struct tagStore *store; |
struct tagSession *sess = (struct tagSession*) arg; |
struct tagSession *sess = (struct tagSession*) arg; |
Line 159 cmdCONNECT(void *srv, void *arg)
|
Line 161 cmdCONNECT(void *srv, void *arg)
|
return -1; |
return -1; |
|
|
ioDEBUG(5, "Exec CONNECT session"); |
ioDEBUG(5, "Exec CONNECT session"); |
SESS_LOCK; |
|
TAILQ_REMOVE(&Sessions, sess, sess_node); |
TAILQ_REMOVE(&Sessions, sess, sess_node); |
SESS_UNLOCK; |
|
|
|
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, "%"); |
|
|
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 177 cmdCONNECT(void *srv, void *arg)
|
Line 176 cmdCONNECT(void *srv, void *arg)
|
|
|
free(store); |
free(store); |
} |
} |
SESS_ELEM_UNLOCK(sess); |
|
|
|
if (sess->sess_will.msg) |
if (sess->sess_will.msg) |
free(sess->sess_will.msg); |
free(sess->sess_will.msg); |
Line 186 cmdCONNECT(void *srv, void *arg)
|
Line 184 cmdCONNECT(void *srv, 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 0; |
return 0; |
} |
} |
|
|
int |
int |
cmdDISCONNECT(void *srv, void *arg) | cmdDISCONNECT(void *srv, int len, void *arg) |
{ |
{ |
struct tagSession *sess = (struct tagSession*) arg; |
struct tagSession *sess = (struct tagSession*) arg; |
|
|
Line 200 cmdDISCONNECT(void *srv, void *arg)
|
Line 200 cmdDISCONNECT(void *srv, void *arg)
|
return -1; |
return -1; |
|
|
ioDEBUG(5, "Exec DISCONNECT session"); |
ioDEBUG(5, "Exec DISCONNECT session"); |
SESS_LOCK; |
|
TAILQ_REMOVE(&Sessions, sess, sess_node); |
|
SESS_UNLOCK; |
|
|
|
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); |
return 0; | |
| return -2; /* must terminate dispatcher */ |
} |
} |