--- mqtt/src/mqttd_calls.c	2012/01/27 15:05:38	1.2
+++ mqtt/src/mqttd_calls.c	2012/01/30 08:29:55	1.2.2.1
@@ -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,127 @@ 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;
+
+	ioTRACE(2);
+
+	if (!sess)
+		return -1;
+
+	hdr = (struct mqtthdr*) sess->sess_buf->msg_base;
+
+	return 0;
+}
+
+int
+cmdCONNECT(void *srv, void *arg)
+{
+	struct tagStore *store;
+	struct tagSession *sess = (struct tagSession*) arg;
+
+	ioTRACE(2);
+
+	if (!sess)
+		return -1;
+
+	pthread_mutex_lock(&mtx_sess);
+	TAILQ_REMOVE(&Sessions, sess, sess_node);
+	pthread_mutex_unlock(&mtx_sess);
+
+	if (call.FiniSessPUB)
+		call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");
+
+	while ((store = TAILQ_FIRST(&sess->sess_sndqueue))) {
+		TAILQ_REMOVE(&sess->sess_sndqueue, store, 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 0;
+}
+
+int
+cmdDISCONNECT(void *srv, void *arg)
+{
+	struct tagSession *sess = (struct tagSession*) arg;
+
+	ioTRACE(2);
+
+	if (!sess)
+		return -1;
+
+	pthread_mutex_lock(&mtx_sess);
+	TAILQ_REMOVE(&Sessions, sess, sess_node);
+	pthread_mutex_unlock(&mtx_sess);
+
+	call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, 
+			sess->sess_addr, sess->sess_user);
 	return 0;
 }