--- mqtt/src/pubmqtt.c	2012/06/26 08:05:58	1.2.2.11
+++ mqtt/src/pubmqtt.c	2012/07/03 09:02:50	1.3
@@ -247,12 +247,14 @@ mqtt_rtlm_chk_session(cfg_root_t *cfg, sqlite3 *sql, c
  * @txtlen = text length
  * @user = username
  * @host = hostname
+ * @qos = QoS
  * @retain = !=0 retain message to database
  * return: -1 error, 0 no publish or >0 published ok
  */
 int
 mqtt_rtlm_write_topic(cfg_root_t *cfg, sqlite3 *sql, const char *connid, u_short msgid, 
-		const char *topic, void *txt, int txtlen, const char *user, const char *host, char retain)
+		const char *topic, void *txt, int txtlen, const char *user, 
+		const char *host, char qos, char retain)
 {
 	int ret = 0;
 	char *str, *psStmt;
@@ -266,12 +268,12 @@ mqtt_rtlm_write_topic(cfg_root_t *cfg, sqlite3 *sql, c
 		mqtt_rtlm_log("Error:: not found topics table name");
 		return -1;
 	}
-	psStmt = sqlite3_mprintf("INSERT INTO %s (Retain, ConnID, MsgID, Topic, Value, PubUser, "
-			"PubDate, PubHost) VALUES (%d, '%q', %u, '%q', ?1, '%q', "
+	psStmt = sqlite3_mprintf("INSERT INTO %s (QoS, Retain, ConnID, MsgID, Topic, Value, PubUser, "
+			"PubDate, PubHost) VALUES (%d, %d, '%q', %u, '%q', ?1, '%q', "
 			"datetime('now', 'localtime'), '%q');", 
-			str, retain, connid, msgid, topic, txt, user, host);
+			str, qos, retain, connid, msgid, topic, user, host);
 
-	if (sqlite3_prepare_v2(sql, psStmt, strlen(psStmt), &stmt, NULL)) {
+	if (sqlite3_prepare_v2(sql, psStmt, strlen(psStmt), &stmt, NULL) || !stmt) {
 		MQTT_RTLM_LOG(sql);
 		sqlite3_free(psStmt);
 		return -1;
@@ -419,13 +421,12 @@ mqtt_rtlm_delete_topic(cfg_root_t *cfg, sqlite3 *sql, 
  * @cfg = loaded config
  * @sql = SQL handle
  * @connid = connection id
- * @msgid = MessageID
  * @topic = topic
  * @retain = retain 0 get only dynamic, >0 get only retained and -1 no matter
  * return: NULL error or not found and !=NULL allocated subscribe topics
  */
 mqtt_subscr_t *
-mqtt_rtlm_read_topic(cfg_root_t *cfg, sqlite3 *sql, const char *connid, u_short msgid, 
+mqtt_rtlm_read_topic(cfg_root_t *cfg, sqlite3 *sql, const char *connid, 
 		const char *topic, char retain)
 {
 	int rowz = 0;
@@ -455,9 +456,9 @@ mqtt_rtlm_read_topic(cfg_root_t *cfg, sqlite3 *sql, co
 		mqtt_rtlm_log("Error:: not found topics table name");
 		return NULL;
 	}
-	psStmt = sqlite3_mprintf("SELECT Retain, Topic, Value FROM %s WHERE "
-			"ConnID = '%q' AND MsgID = %d AND Topic LIKE '%q' %s;", 
-			str, connid, msgid, topic, szStr);
+	psStmt = sqlite3_mprintf("SELECT QoS, Topic, Value  FROM %s WHERE "
+			"ConnID LIKE '%q' AND Topic LIKE '%q' %s;", 
+			str, connid, topic, szStr);
 
 	if (sqlite3_prepare_v2(sql, psStmt, strlen(psStmt), &stmt, NULL)) {
 		MQTT_RTLM_LOG(sql);
@@ -469,7 +470,7 @@ mqtt_rtlm_read_topic(cfg_root_t *cfg, sqlite3 *sql, co
 	/* calculate count of rows and allocate subscribe items */
 	while (sqlite3_step(stmt) == SQLITE_ROW)
 		rowz++;
-	if (!(s = io_malloc((rowz + 1) * sizeof(mqtt_subscr_t)))) {
+	if (!(s = malloc((rowz + 1) * sizeof(mqtt_subscr_t)))) {
 		mqtt_rtlm_log("Error:: System #%d - %s", errno, strerror(errno));
 		goto end;
 	} else
@@ -479,13 +480,14 @@ mqtt_rtlm_read_topic(cfg_root_t *cfg, sqlite3 *sql, co
 	/* fill with data */
 	for (j = 0; j < rowz && sqlite3_step(stmt) == SQLITE_ROW; j++) {
 		s[j].sub_ret = (char) sqlite3_column_int(stmt, 0);
-		s[j].sub_topic.msg_base = (u_char*) io_strdup((char*) sqlite3_column_text(stmt, 1));
+		s[j].sub_topic.msg_base = (u_char*) strdup((char*) sqlite3_column_text(stmt, 1));
 		s[j].sub_topic.msg_len = strlen((char*) s[j].sub_topic.msg_base);
 		AIT_SET_PTR(&v, (void*) sqlite3_column_blob(stmt, 2), sqlite3_column_bytes(stmt, 2));
 		s[j].sub_value.msg_len = AIT_LEN(&v);
-		s[j].sub_value.msg_base = (u_char*) io_malloc(s[j].sub_value.msg_len);
+		s[j].sub_value.msg_base = (u_char*) malloc(s[j].sub_value.msg_len);
 		if (s[j].sub_value.msg_base)
 			memcpy(s[j].sub_value.msg_base, AIT_GET_PTR(&v), s[j].sub_value.msg_len);
+		AIT_FREE_VAL(&v);
 	}
 end:
 	sqlite3_finalize(stmt);
@@ -633,7 +635,7 @@ mqtt_rtlm_read_subscribe(cfg_root_t *cfg, sqlite3 *sql
 	/* calculate count of rows and allocate subscribe items */
 	while (sqlite3_step(stmt) == SQLITE_ROW)
 		rowz++;
-	if (!(s = io_malloc((rowz + 1) * sizeof(mqtt_subscr_t)))) {
+	if (!(s = malloc((rowz + 1) * sizeof(mqtt_subscr_t)))) {
 		mqtt_rtlm_log("Error:: System #%d - %s", errno, strerror(errno));
 		goto end;
 	} else
@@ -643,7 +645,7 @@ mqtt_rtlm_read_subscribe(cfg_root_t *cfg, sqlite3 *sql
 	/* fill with data */
 	for (j = 0; j < rowz && sqlite3_step(stmt) == SQLITE_ROW; j++) {
 		s[j].sub_ret = (char) sqlite3_column_int(stmt, 0);
-		s[j].sub_topic.msg_base = (u_char*) io_strdup((char*) sqlite3_column_text(stmt, 1));
+		s[j].sub_topic.msg_base = (u_char*) strdup((char*) sqlite3_column_text(stmt, 1));
 		s[j].sub_topic.msg_len = strlen((char*) s[j].sub_topic.msg_base);
 		s[j].sub_value.msg_base = NULL;
 		s[j].sub_value.msg_len = 0;