#include "global.h" #include "mqttd.h" #include "rtlm.h" #include "mqttd_calls.h" int cmdPUBLISH(void *srv, void *arg) { struct mqtthdr *hdr; struct tagSession *sess = (struct tagSession*) arg; ioTRACE(2); if (!sess) return -1; hdr = (struct mqtthdr*) sess->sess_buf->msg_base; switch (hdr->mqtt_msg.qos) { case MQTT_QOS_ONCE: break; case MQTT_QOS_ACK: break; case MQTT_QOS_EXACTLY: break; default: ioDEBUG(1, "Error:: Unknown QoS %d - rejected publishing request", hdr->mqtt_msg.qos); return 0; } return 0; } int cmdPUBREL(void *srv, void *arg) { struct mqtthdr *hdr; struct tagSession *sess = (struct tagSession*) arg; ioTRACE(2); if (!sess) return -1; hdr = (struct mqtthdr*) sess->sess_buf->msg_base; return 0; } int cmdSUBSCRIBE(void *srv, void *arg) { struct mqtthdr *hdr; struct tagSession *sess = (struct tagSession*) arg; ioTRACE(2); if (!sess) return -1; hdr = (struct mqtthdr*) sess->sess_buf->msg_base; return 0; } int cmdUNSUBSCRIBE(void *srv, void *arg) { struct mqtthdr *hdr; struct tagSession *sess = (struct tagSession*) arg; ioTRACE(2); if (!sess) return -1; hdr = (struct mqtthdr*) sess->sess_buf->msg_base; return 0; } int cmdPINGREQ(void *srv, void *arg) { struct mqtthdr *hdr; struct tagSession *sess = (struct tagSession*) arg; int siz = 0; ioTRACE(2); if (!sess) return -1; hdr = (struct mqtthdr*) sess->sess_buf->msg_base; siz = mqtt_msgPINGRESP(sess->sess_buf); if (siz == -1) { ioDEBUG(5, "Error:: in msgPINGRESP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, 0)) == -1) { ioSYSERR(0); return 0; } else ioDEBUG(5, "Sended %d bytes.", siz); return 0; } int cmdCONNECT(void *srv, void *arg) { struct tagStore *store; struct tagSession *sess = (struct tagSession*) arg; register int i; ioTRACE(2); if (!sess) return -1; SESS_LOCK; TAILQ_REMOVE(&Sessions, sess, sess_node); SESS_UNLOCK; if (call.FiniSessPUB) call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); SESS_ELEM_LOCK(sess); for (i = 0; i < MQTT_QOS_RESERVED; i++) while ((store = SLIST_FIRST(&sess->sess_txque[i]))) { SLIST_REMOVE_HEAD(&sess->sess_txque[i], 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); } SESS_ELEM_UNLOCK(sess); if (sess->sess_will.msg) free(sess->sess_will.msg); if (sess->sess_will.topic) free(sess->sess_will.topic); call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, sess->sess_addr, sess->sess_user); return 0; } int cmdDISCONNECT(void *srv, void *arg) { struct tagSession *sess = (struct tagSession*) arg; ioTRACE(2); if (!sess) return -1; 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, sess->sess_addr, sess->sess_user); return 0; }