#include "global.h" #include "mqttd.h" #include "rtlm.h" #include "mqttd_calls.h" int cmdPUBLISH(void *srv, int len, 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, int len, 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, int len, void *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; ioTRACE(2); if (!sess) return -1; ioDEBUG(5, "Exec SUBSCRIBE session"); siz = mqtt_readSUBSCRIBE(sess->sess_buf, &mid, &subs); if (siz == -1) { ioDEBUG(5, "Error:: in readSUBSCRIBE #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } /* add to db */ for (i = 0; i < siz; i++) { 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; mqtt_subCopy(&store->st_subscr, &subs[i]); } /* add to cache */ SLIST_INSERT_HEAD(&sess->sess_subscr, store, st_node); subs[i].sub_ret = MQTT_QOS_PASS; } else subs[i].sub_ret = MQTT_QOS_DENY; } /* send acknowledge */ siz = mqtt_msgSUBACK(sess->sess_buf, subs, mid); if (siz == -1) { ioDEBUG(5, "Error:: in msgSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); goto end; } if ((siz = send(sess->sess_sock, sess->sess_buf->msg_base, siz, 0)) == -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; } int cmdUNSUBSCRIBE(void *srv, int len, void *arg) { struct tagSession *sess = (struct tagSession*) arg; ioTRACE(2); if (!sess) return -1; return 0; } int cmdPINGREQ(void *srv, int len, void *arg) { struct tagSession *sess = (struct tagSession*) arg; int siz = 0; ioTRACE(2); if (!sess) return -1; ioDEBUG(5, "Exec PINGREQ session"); 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); memset(sess->sess_buf->msg_base, 0, sess->sess_buf->msg_len); } return 0; } int cmdCONNECT(void *srv, int len, void *arg) { struct tagStore *store; struct tagSession *sess = (struct tagSession*) arg; ioTRACE(2); if (!sess) return -1; ioDEBUG(5, "Exec CONNECT session"); TAILQ_REMOVE(&Sessions, sess, sess_node); if (call.FiniSessPUB) call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%"); while ((store = SLIST_FIRST(&sess->sess_subscr))) { SLIST_REMOVE_HEAD(&sess->sess_subscr, 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); } 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 -3; /* reconnect client */ } int cmdDISCONNECT(void *srv, int len, void *arg) { struct tagSession *sess = (struct tagSession*) arg; ioTRACE(2); if (!sess) return -1; ioDEBUG(5, "Exec DISCONNECT session"); call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, sess->sess_addr, sess->sess_user); return -2; /* must terminate dispatcher */ }