--- mqtt/src/mqttd_calls.c	2012/01/27 15:05:38	1.2
+++ mqtt/src/mqttd_calls.c	2012/04/11 16:37:17	1.2.2.3
@@ -1,12 +1,14 @@
 #include "global.h"
 #include "mqttd.h"
+#include "rtlm.h"
 #include "mqttd_calls.h"
 
 
 int
-Publish(struct tagSession *sess)
+cmdPUBLISH(void *srv, void *arg)
 {
 	struct mqtthdr *hdr;
+	struct tagSession *sess = (struct tagSession*) arg;
 
 	ioTRACE(2);
 
@@ -27,5 +29,142 @@ Publish(struct tagSession *sess)
 			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;
 }