--- mqtt/src/pubmqtt.c 2011/11/28 22:31:19 1.1.2.13 +++ mqtt/src/pubmqtt.c 2012/07/03 12:22:56 1.3.2.1 @@ -1,3 +1,48 @@ +/************************************************************************* +* (C) 2011 AITNET ltd - Sofia/Bulgaria - +* by Michael Pounov +* +* $Author: misho $ +* $Id: pubmqtt.c,v 1.3.2.1 2012/07/03 12:22:56 misho Exp $ +* +************************************************************************** +The ELWIX and AITNET software is distributed under the following +terms: + +All of the documentation and software included in the ELWIX and AITNET +Releases is copyrighted by ELWIX - Sofia/Bulgaria + +Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012 + by Michael Pounov . All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions +are met: +1. Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. +2. Redistributions in binary form must reproduce the above copyright + notice, this list of conditions and the following disclaimer in the + documentation and/or other materials provided with the distribution. +3. All advertising materials mentioning features or use of this software + must display the following acknowledgement: +This product includes software developed by Michael Pounov +ELWIX - Embedded LightWeight unIX and its contributors. +4. Neither the name of AITNET nor the names of its contributors + may be used to endorse or promote products derived from this software + without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY AITNET AND CONTRIBUTORS ``AS IS'' AND +ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE +FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS +OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) +HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT +LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY +OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF +SUCH DAMAGE. +*/ #include "global.h" @@ -20,10 +65,24 @@ mqtt_rtlm_log(const char *fmt, ...) vsyslog(LOG_ERR, fmt, lst); va_end(lst); } -#define MQTT_RTLM_LOG(_sql) (assert((_sql)), mqtt_rtlm_log("Error:: SQL #%d - %s", \ +#define MQTT_RTLM_LOG(_sql) (assert((_sql)), mqtt_rtlm_log("Error:: %s(%d) SQL #%d - %s", \ + __func__, __LINE__, \ sqlite3_errcode((_sql)), sqlite3_errmsg((_sql)))) +/* library pre-loaded actions */ +void +_init() +{ + sqlite3_initialize(); +} +void +_fini() +{ + sqlite3_shutdown(); +} + + /* * mqtt_rtlm_open() Open database connection * @@ -31,7 +90,7 @@ mqtt_rtlm_log(const char *fmt, ...) * return: NULL error or SQL handle */ sqlite3 * -mqtt_rtlm_open(sl_config *cfg) +mqtt_rtlm_open(cfg_root_t *cfg) { sqlite3 *sql = NULL; const char *str = NULL; @@ -39,9 +98,7 @@ mqtt_rtlm_open(sl_config *cfg) if (!cfg) return NULL; - sqlite3_config(SQLITE_CONFIG_SERIALIZED); - - str = (const char*) cfg_GetAttribute(cfg, CFG("mqtt_pub"), CFG("name")); + str = cfg_getAttribute(cfg, "mqtt_pub", "name"); if (!str) { mqtt_rtlm_log("Error:: Unknown database name ...\n"); return NULL; @@ -78,33 +135,49 @@ mqtt_rtlm_close(sqlite3 *sql) * * @cfg = loaded config * @sql = SQL handle + * @connid = connection id * @user = username * @host = hostname - * @port = port + * @will = will flag if !=0 must fill arguments + * @... = will arguments in order topic,msg,qos,retain * return: -1 error, 0 session already appears or >0 row changed */ int -mqtt_rtlm_init_session(sl_config *cfg, sqlite3 *sql, const char *user, const char *host, u_short port) +mqtt_rtlm_init_session(cfg_root_t *cfg, sqlite3 *sql, const char *connid, const char *user, + const char *host, char will, ...) { + va_list lst; int ret = 0; - char *str, szStmt[BUFSIZ] = { 0 }; + char *str, *psStmt; sqlite3_stmt *stmt; if (!cfg || !sql) return -1; - str = (char*) cfg_GetAttribute(cfg, CFG("mqtt_pub"), CFG("tbl_online")); + str = (char*) cfg_getAttribute(cfg, "mqtt_pub", "tbl_online"); if (!str) { mqtt_rtlm_log("Error:: not found online table name"); return -1; } - snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (Username, RemoteHost, RemotePort) " - "VALUES ('%s', '%s', %d);", str, user, host, port); + if (!will) + psStmt = sqlite3_mprintf("INSERT INTO %s (ConnID, Username, RemoteHost, " + "WillFlag) VALUES ('%q', '%q', '%q', 0);", str, connid, user, host); + else { + va_start(lst, will); + psStmt = sqlite3_mprintf("INSERT INTO %s (ConnID, Username, RemoteHost, " + "WillFlag, WillRetain, WillQoS, WillMsg, WillTopic) " + "VALUES ('%q', '%q', '%q', %d, %d, %d, '%q', '%q');", + str, connid, user, host, will, + va_arg(lst, int), va_arg(lst, int), va_arg(lst, char*), va_arg(lst, char*)); + va_end(lst); + } - if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) { + if (sqlite3_prepare_v2(sql, psStmt, strlen(psStmt), &stmt, NULL)) { MQTT_RTLM_LOG(sql); + sqlite3_free(psStmt); return -1; - } + } else + sqlite3_free(psStmt); if ((ret = sqlite3_step(stmt)) == SQLITE_DONE) ret = sqlite3_changes(sql); else { @@ -122,32 +195,35 @@ mqtt_rtlm_init_session(sl_config *cfg, sqlite3 *sql, c * * @cfg = loaded config * @sql = SQL handle + * @connid = connection id * @user = username * @host = hostname * return: -1 error, 0 session already appears or >0 row changed */ int -mqtt_rtlm_fini_session(sl_config *cfg, sqlite3 *sql, const char *user, const char *host) +mqtt_rtlm_fini_session(cfg_root_t *cfg, sqlite3 *sql, const char *connid, const char *user, const char *host) { int ret = 0; - char *str, szStmt[BUFSIZ] = { 0 }; + char *str, *psStmt; sqlite3_stmt *stmt; if (!cfg || !sql) return -1; - str = (char*) cfg_GetAttribute(cfg, CFG("mqtt_pub"), CFG("tbl_online")); + str = (char*) cfg_getAttribute(cfg, "mqtt_pub", "tbl_online"); if (!str) { mqtt_rtlm_log("Error:: not found online table name"); return -1; } - snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE Username = '%s' AND RemoteHost LIKE '%s';", - str, user, host); + psStmt = sqlite3_mprintf("DELETE FROM %s WHERE ConnID = '%q' AND Username = '%q' " + "AND RemoteHost LIKE '%q';", str, connid, user, host); - if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) { + if (sqlite3_prepare_v2(sql, psStmt, strlen(psStmt), &stmt, NULL)) { MQTT_RTLM_LOG(sql); + sqlite3_free(psStmt); return -1; - } + } else + sqlite3_free(psStmt); if ((ret = sqlite3_step(stmt)) == SQLITE_DONE) ret = sqlite3_changes(sql); else { @@ -165,32 +241,36 @@ mqtt_rtlm_fini_session(sl_config *cfg, sqlite3 *sql, c * * @cfg = loaded config * @sql = SQL handle + * @connid = connection id * @user = username * @host = hostname * return: -1 error, 0 not logged or >0 logged found rows */ int -mqtt_rtlm_chk_session(sl_config *cfg, sqlite3 *sql, const char *user, const char *host) +mqtt_rtlm_chk_session(cfg_root_t *cfg, sqlite3 *sql, const char *connid, const char *user, const char *host) { int ret = 0; - char *str, szStmt[BUFSIZ] = { 0 }; + char *str, *psStmt; sqlite3_stmt *stmt; if (!cfg || !sql) return -1; - str = (char*) cfg_GetAttribute(cfg, CFG("mqtt_pub"), CFG("tbl_online")); + str = (char*) cfg_getAttribute(cfg, "mqtt_pub", "tbl_online"); if (!str) { mqtt_rtlm_log("Error:: not found online table name"); return -1; } - snprintf(szStmt, sizeof szStmt, "SELECT RemoteHost, RemotePort FROM %s WHERE " - "Username = '%s' AND RemoteHost LIKE '%s';", str, user, host); + psStmt = sqlite3_mprintf("SELECT ConnID, RemoteHost FROM %s WHERE " + "ConnID = '%q' AND Username LIKE '%q' AND RemoteHost LIKE '%q';", + str, connid, user, host); - if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) { + if (sqlite3_prepare_v2(sql, psStmt, strlen(psStmt), &stmt, NULL)) { MQTT_RTLM_LOG(sql); + sqlite3_free(psStmt); return -1; - } + } else + sqlite3_free(psStmt); if (sqlite3_step(stmt) == SQLITE_ROW) ret = sqlite3_changes(sql); else @@ -205,36 +285,49 @@ mqtt_rtlm_chk_session(sl_config *cfg, sqlite3 *sql, co * * @cfg = loaded config * @sql = SQL handle + * @connid = connection id + * @msgid = MessageID * @topic = topic * @txt = text + * @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(sl_config *cfg, sqlite3 *sql, const char *topic, const char *txt, - const char *user, const char *host, char retain) +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 qos, char retain) { int ret = 0; - char *str, szStmt[BUFSIZ] = { 0 }; + char *str, *psStmt; sqlite3_stmt *stmt; if (!cfg || !sql || !topic) return -1; - str = (char*) cfg_GetAttribute(cfg, CFG("mqtt_pub"), CFG("tbl_topics")); + str = (char*) cfg_getAttribute(cfg, "mqtt_pub", "tbl_topics"); if (!str) { mqtt_rtlm_log("Error:: not found topics table name"); return -1; } - snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (Retain, Topic, Value, PubUser, PubDate, PubHost) " - "VALUES (%d, '%s', '%s', '%s', datetime('now', 'localtime'), '%s');", str, - retain, topic, txt, user, host); + 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, qos, retain, connid, msgid, topic, user, host); - if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) { + if (sqlite3_prepare_v2(sql, psStmt, strlen(psStmt), &stmt, NULL) || !stmt) { MQTT_RTLM_LOG(sql); + sqlite3_free(psStmt); return -1; + } else + sqlite3_free(psStmt); + if (sqlite3_bind_blob(stmt, 1, txt, txtlen, SQLITE_TRANSIENT)) { + MQTT_RTLM_LOG(sql); + sqlite3_finalize(stmt); + return -1; } if ((ret = sqlite3_step(stmt)) == SQLITE_DONE) ret = sqlite3_changes(sql); @@ -249,10 +342,69 @@ mqtt_rtlm_write_topic(sl_config *cfg, sqlite3 *sql, co } /* + * mqtt_rtlm_wipe_topic() Wipe all topics + * + * @cfg = loaded config + * @sql = SQL handle + * @connid = connection id + * @user = username + * @retain = -1 no matter + * return: -1 error, 0 no changes or >0 deleted rows + */ +int +mqtt_rtlm_wipe_topic(cfg_root_t *cfg, sqlite3 *sql, const char *connid, const char *user, char retain) +{ + int ret = 0; + char *str, *rtn, *psStmt; + sqlite3_stmt *stmt; + + if (!cfg || !sql || !connid) + return -1; + + str = (char*) cfg_getAttribute(cfg, "mqtt_pub", "tbl_topics"); + if (!str) { + mqtt_rtlm_log("Error:: not found topics table name"); + return -1; + } + switch (retain) { + case -1: + rtn = ""; + break; + case 0: + rtn = "AND Retain = 0"; + break; + default: + rtn = "AND Retain != 0"; + break; + } + psStmt = sqlite3_mprintf("DELETE FROM %s WHERE ConnID = '%q' AND " + "PubUser LIKE '%q' %s;", str, connid, user, rtn); + + if (sqlite3_prepare_v2(sql, psStmt, strlen(psStmt), &stmt, NULL)) { + MQTT_RTLM_LOG(sql); + sqlite3_free(psStmt); + return -1; + } else + sqlite3_free(psStmt); + if ((ret = sqlite3_step(stmt)) == SQLITE_DONE) + ret = sqlite3_changes(sql); + else { + if (ret > SQLITE_OK && ret < SQLITE_ROW) + MQTT_RTLM_LOG(sql); + ret = 0; + } + sqlite3_finalize(stmt); + + return ret; +} + +/* * mqtt_rtlm_delete_topic() Delete topic * * @cfg = loaded config * @sql = SQL handle + * @connid = connection id + * @msgid = MessageID * @topic = topic * @user = username * @host = hostname @@ -260,17 +412,17 @@ mqtt_rtlm_write_topic(sl_config *cfg, sqlite3 *sql, co * return: -1 error, 0 no changes or >0 deleted rows */ int -mqtt_rtlm_delete_topic(sl_config *cfg, sqlite3 *sql, const char *topic, - const char *user, const char *host, char retain) +mqtt_rtlm_delete_topic(cfg_root_t *cfg, sqlite3 *sql, const char *connid, u_short msgid, + const char *topic, const char *user, const char *host, char retain) { int ret = 0; - char *str, *rtn, szStmt[BUFSIZ] = { 0 }; + char *str, *rtn, *psStmt; sqlite3_stmt *stmt; if (!cfg || !sql || !topic) return -1; - str = (char*) cfg_GetAttribute(cfg, CFG("mqtt_pub"), CFG("tbl_topics")); + str = (char*) cfg_getAttribute(cfg, "mqtt_pub", "tbl_topics"); if (!str) { mqtt_rtlm_log("Error:: not found topics table name"); return -1; @@ -286,14 +438,16 @@ mqtt_rtlm_delete_topic(sl_config *cfg, sqlite3 *sql, c rtn = "AND Retain != 0"; break; } - snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE Topic LIKE '%s' AND " - "PubUser LIKE '%s' AND PubHost LIKE '%s' %s;", str, - topic, user, host, rtn); + psStmt = sqlite3_mprintf("DELETE FROM %s WHERE ConnID = '%q' AND MsgID = %d AND " + "Topic LIKE '%q' AND PubUser LIKE '%q' AND PubHost LIKE '%q' %s;", str, + connid, msgid, topic, user, host, rtn); - if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) { + if (sqlite3_prepare_v2(sql, psStmt, strlen(psStmt), &stmt, NULL)) { MQTT_RTLM_LOG(sql); + sqlite3_free(psStmt); return -1; - } + } else + sqlite3_free(psStmt); if ((ret = sqlite3_step(stmt)) == SQLITE_DONE) ret = sqlite3_changes(sql); else { @@ -311,18 +465,21 @@ mqtt_rtlm_delete_topic(sl_config *cfg, sqlite3 *sql, c * * @cfg = loaded config * @sql = SQL handle + * @connid = connection id * @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(sl_config *cfg, sqlite3 *sql, const char *topic, char retain) +mqtt_rtlm_read_topic(cfg_root_t *cfg, sqlite3 *sql, const char *connid, + const char *topic, char retain) { int rowz = 0; - char *str, szStr[STRSIZ], szStmt[BUFSIZ] = { 0 }; + char *str, szStr[STRSIZ], *psStmt; sqlite3_stmt *stmt; register int j; mqtt_subscr_t *s = NULL; + ait_val_t v; if (!cfg || !sql || !topic) return NULL; @@ -339,19 +496,187 @@ mqtt_rtlm_read_topic(sl_config *cfg, sqlite3 *sql, con break; } - str = (char*) cfg_GetAttribute(cfg, CFG("mqtt_pub"), CFG("tbl_topics")); + str = (char*) cfg_getAttribute(cfg, "mqtt_pub", "tbl_topics"); if (!str) { mqtt_rtlm_log("Error:: not found topics table name"); return NULL; } - snprintf(szStmt, sizeof szStmt, "SELECT Retain, Topic, Value FROM %s WHERE Topic LIKE '%s' %s;", - str, 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, szStmt, strlen(szStmt), &stmt, NULL)) { + if (sqlite3_prepare_v2(sql, psStmt, strlen(psStmt), &stmt, NULL)) { MQTT_RTLM_LOG(sql); + sqlite3_free(psStmt); return NULL; + } else + sqlite3_free(psStmt); + + /* calculate count of rows and allocate subscribe items */ + while (sqlite3_step(stmt) == SQLITE_ROW) + rowz++; + if (!(s = malloc((rowz + 1) * sizeof(mqtt_subscr_t)))) { + mqtt_rtlm_log("Error:: System #%d - %s", errno, strerror(errno)); + goto end; + } else + memset(s, 0, (rowz + 1) * sizeof(mqtt_subscr_t)); + sqlite3_reset(stmt); + + /* 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*) 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*) 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); + return s; +} + +/* + * mqtt_rtlm_write_subscribe() Subscribe topic + * + * @cfg = loaded config + * @sql = SQL handle + * @connid = connection id + * @msgid = MessageID + * @topic = topic + * @user = username + * @host = hostname + * @qos = Subscribe QoS + * return: -1 error, 0 no publish or >0 published ok + */ +int +mqtt_rtlm_write_subscribe(cfg_root_t *cfg, sqlite3 *sql, const char *connid, u_short msgid, + const char *topic, const char *user, const char *host, char qos) +{ + int ret = 0; + char *str, *psStmt; + sqlite3_stmt *stmt; + + if (!cfg || !sql || !topic) + return -1; + + str = (char*) cfg_getAttribute(cfg, "mqtt_pub", "tbl_subscribes"); + if (!str) { + mqtt_rtlm_log("Error:: not found subscribes table name"); + return -1; + } + psStmt = sqlite3_mprintf("INSERT INTO %s (ConnID, MsgID, QoS, Topic, PubUser, " + "PubDate, PubHost) VALUES ('%q', %d, %d, '%q', '%q', " + "datetime('now', 'localtime'), '%q');", str, + connid, msgid, qos, topic, user, host); + + if (sqlite3_prepare_v2(sql, psStmt, strlen(psStmt), &stmt, NULL)) { + MQTT_RTLM_LOG(sql); + sqlite3_free(psStmt); + return -1; + } else + sqlite3_free(psStmt); + if ((ret = sqlite3_step(stmt)) == SQLITE_DONE) + ret = sqlite3_changes(sql); + else { + if (ret > SQLITE_OK && ret < SQLITE_ROW) + MQTT_RTLM_LOG(sql); + ret = 0; + } + sqlite3_finalize(stmt); + + return ret; +} + +/* + * mqtt_rtlm_delete_subscribe() Delete subscribe + * + * @cfg = loaded config + * @sql = SQL handle + * @connid = connection id + * @topic = topic + * @user = username + * @host = hostname + * return: -1 error, 0 no changes or >0 deleted rows + */ +int +mqtt_rtlm_delete_subscribe(cfg_root_t *cfg, sqlite3 *sql, const char *connid, + const char *topic, const char *user, const char *host) +{ + int ret = 0; + char *str, *psStmt; + sqlite3_stmt *stmt; + + if (!cfg || !sql || !topic) + return -1; + + str = (char*) cfg_getAttribute(cfg, "mqtt_pub", "tbl_subscribes"); + if (!str) { + mqtt_rtlm_log("Error:: not found subscribes table name"); + return -1; + } + psStmt = sqlite3_mprintf("DELETE FROM %s WHERE ConnID = '%q' AND " + "Topic LIKE '%q' AND PubUser LIKE '%q' AND PubHost LIKE '%q';", str, + connid, topic, user, host); + + if (sqlite3_prepare_v2(sql, psStmt, strlen(psStmt), &stmt, NULL)) { + MQTT_RTLM_LOG(sql); + sqlite3_free(psStmt); + return -1; + } else + sqlite3_free(psStmt); + if ((ret = sqlite3_step(stmt)) == SQLITE_DONE) + ret = sqlite3_changes(sql); + else { + if (ret > SQLITE_OK && ret < SQLITE_ROW) + MQTT_RTLM_LOG(sql); + ret = 0; + } + sqlite3_finalize(stmt); + + return ret; +} + +/* + * mqtt_rtlm_read_subscribe() Get subscribe topic + * + * @cfg = loaded config + * @sql = SQL handle + * @connid = connection id + * @topic = topic + * return: NULL error or not found and !=NULL allocated subscribe topics + */ +mqtt_subscr_t * +mqtt_rtlm_read_subscribe(cfg_root_t *cfg, sqlite3 *sql, const char *connid, const char *topic) +{ + int rowz = 0; + char *str, *psStmt; + sqlite3_stmt *stmt; + register int j; + mqtt_subscr_t *s = NULL; + + if (!cfg || !sql || !topic) + return NULL; + + str = (char*) cfg_getAttribute(cfg, "mqtt_pub", "tbl_subscribes"); + if (!str) { + mqtt_rtlm_log("Error:: not found subscribes table name"); + return NULL; + } + psStmt = sqlite3_mprintf("SELECT QoS, Topic FROM %s WHERE ConnID = '%q' AND " + "Topic LIKE '%q';", str, connid, topic); + + if (sqlite3_prepare_v2(sql, psStmt, strlen(psStmt), &stmt, NULL)) { + MQTT_RTLM_LOG(sql); + sqlite3_free(psStmt); + return NULL; + } else + sqlite3_free(psStmt); + /* calculate count of rows and allocate subscribe items */ while (sqlite3_step(stmt) == SQLITE_ROW) rowz++; @@ -365,10 +690,10 @@ mqtt_rtlm_read_topic(sl_config *cfg, sqlite3 *sql, con /* 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._base = strdup(sqlite3_column_text(stmt, 1)); - s[j].sub_topic._size = strlen(s[j].sub_topic._base); - s[j].sub_value._base = strdup(sqlite3_column_text(stmt, 2)); - s[j].sub_value._size = strlen(s[j].sub_value._base); + 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; } end: sqlite3_finalize(stmt);