--- mqtt/src/mqttd_calls.c 2012/04/27 10:07:51 1.2.2.7 +++ mqtt/src/mqttd_calls.c 2012/05/05 13:39:27 1.2.2.9 @@ -5,7 +5,7 @@ int -cmdPUBLISH(void *srv, void *arg) +cmdPUBLISH(void *srv, int len, void *arg) { struct mqtthdr *hdr; struct tagSession *sess = (struct tagSession*) arg; @@ -33,7 +33,7 @@ cmdPUBLISH(void *srv, void *arg) } int -cmdPUBREL(void *srv, void *arg) +cmdPUBREL(void *srv, int len, void *arg) { struct mqtthdr *hdr; struct tagSession *sess = (struct tagSession*) arg; @@ -49,7 +49,7 @@ cmdPUBREL(void *srv, void *arg) } int -cmdSUBSCRIBE(void *srv, void *arg) +cmdSUBSCRIBE(void *srv, int len, void *arg) { struct tagSession *sess = (struct tagSession*) arg; mqtt_subscr_t *subs = NULL; @@ -72,15 +72,15 @@ cmdSUBSCRIBE(void *srv, void *arg) /* add to db */ for (i = 0; i < siz; i++) { - if ((siz = call.WritePUB_subscribe(&cfg, pub, mid, subs[i].sub_topic.msg_base, - sess->sess_user, sess->sess_addr, subs[i].sub_ret)) > 0) { + if (call.WritePUB_subscribe(&cfg, pub, mid, subs[i].sub_topic.msg_base, + sess->sess_user, sess->sess_addr, subs[i].sub_ret) > 0) { store = malloc(sizeof(struct tagStore)); if (!store) { ioSYSERR(0); goto end; } else { store->st_msgid = mid; - store->st_subscr = subs[i]; + mqtt_subCopy(&store->st_subscr, &subs[i]); } /* add to cache */ @@ -101,15 +101,17 @@ cmdSUBSCRIBE(void *srv, void *arg) } if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, 0)) == -1) ioSYSERR(0); - else + 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; } int -cmdUNSUBSCRIBE(void *srv, void *arg) +cmdUNSUBSCRIBE(void *srv, int len, void *arg) { struct tagSession *sess = (struct tagSession*) arg; @@ -122,7 +124,7 @@ cmdUNSUBSCRIBE(void *srv, void *arg) } int -cmdPINGREQ(void *srv, void *arg) +cmdPINGREQ(void *srv, int len, void *arg) { struct tagSession *sess = (struct tagSession*) arg; int siz = 0; @@ -141,14 +143,16 @@ cmdPINGREQ(void *srv, void *arg) if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, 0)) == -1) { ioSYSERR(0); return 0; - } else + } else { ioDEBUG(5, "Sended %d bytes.", siz); + memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); + } return 0; } int -cmdCONNECT(void *srv, void *arg) +cmdCONNECT(void *srv, int len, void *arg) { struct tagStore *store; struct tagSession *sess = (struct tagSession*) arg; @@ -186,11 +190,13 @@ cmdCONNECT(void *srv, void *arg) call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, sess->sess_addr, sess->sess_user); + +// schedEvent(root, startSession, NULL, (u_long) sess->sess_sock, sess, ret); return 0; } int -cmdDISCONNECT(void *srv, void *arg) +cmdDISCONNECT(void *srv, int len, void *arg) { struct tagSession *sess = (struct tagSession*) arg; @@ -206,5 +212,7 @@ cmdDISCONNECT(void *srv, void *arg) call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, sess->sess_addr, sess->sess_user); + + finiSession(sess); return 0; }