--- mqtt/src/pubmqtt.c 2012/05/27 10:12:48 1.2.2.10 +++ mqtt/src/pubmqtt.c 2012/07/03 12:46:01 1.4 @@ -1,3 +1,48 @@ +/************************************************************************* +* (C) 2011 AITNET ltd - Sofia/Bulgaria - +* by Michael Pounov +* +* $Author: misho $ +* $Id: pubmqtt.c,v 1.4 2012/07/03 12:46:01 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" @@ -103,7 +148,7 @@ mqtt_rtlm_init_session(cfg_root_t *cfg, sqlite3 *sql, { va_list lst; int ret = 0; - char *str, szStmt[BUFSIZ] = { 0 }; + char *str, *psStmt; sqlite3_stmt *stmt; if (!cfg || !sql) @@ -115,22 +160,24 @@ mqtt_rtlm_init_session(cfg_root_t *cfg, sqlite3 *sql, return -1; } if (!will) - snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (ConnID, Username, RemoteHost, " - "WillFlag) VALUES ('%s', '%s', '%s', 0);", str, connid, user, host); + psStmt = sqlite3_mprintf("INSERT INTO %s (ConnID, Username, RemoteHost, " + "WillFlag) VALUES ('%q', '%q', '%q', 0);", str, connid, user, host); else { va_start(lst, will); - snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (ConnID, Username, RemoteHost, " + psStmt = sqlite3_mprintf("INSERT INTO %s (ConnID, Username, RemoteHost, " "WillFlag, WillRetain, WillQoS, WillMsg, WillTopic) " - "VALUES ('%s', '%s', '%s', %d, %d, %d, '%s', '%s');", + "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 { @@ -157,7 +204,7 @@ int 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) @@ -168,13 +215,15 @@ mqtt_rtlm_fini_session(cfg_root_t *cfg, sqlite3 *sql, mqtt_rtlm_log("Error:: not found online table name"); return -1; } - snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE ConnID = '%s' AND Username = '%s' " - "AND RemoteHost LIKE '%s';", str, connid, 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 { @@ -201,7 +250,7 @@ int 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) @@ -212,14 +261,16 @@ mqtt_rtlm_chk_session(cfg_root_t *cfg, sqlite3 *sql, c mqtt_rtlm_log("Error:: not found online table name"); return -1; } - snprintf(szStmt, sizeof szStmt, "SELECT ConnID, RemoteHost FROM %s WHERE " - "ConnID = '%s' AND Username LIKE '%s' AND RemoteHost LIKE '%s';", + 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 @@ -238,17 +289,20 @@ mqtt_rtlm_chk_session(cfg_root_t *cfg, sqlite3 *sql, c * @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(cfg_root_t *cfg, sqlite3 *sql, const char *connid, u_short msgid, - const char *topic, const char *txt, 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, szStmt[BUFSIZ] = { 0 }; + char *str, *psStmt; sqlite3_stmt *stmt; if (!cfg || !sql || !topic) @@ -259,14 +313,21 @@ mqtt_rtlm_write_topic(cfg_root_t *cfg, sqlite3 *sql, c mqtt_rtlm_log("Error:: not found topics table name"); return -1; } - snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (Retain, ConnID, MsgID, Topic, Value, PubUser, " - "PubDate, PubHost) VALUES (%d, '%s', %d, '%s', '%s', '%s', " - "datetime('now', 'localtime'), '%s');", - str, retain, connid, msgid, 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); @@ -294,7 +355,7 @@ 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, szStmt[BUFSIZ] = { 0 }; + char *str, *rtn, *psStmt; sqlite3_stmt *stmt; if (!cfg || !sql || !connid) @@ -316,13 +377,15 @@ mqtt_rtlm_wipe_topic(cfg_root_t *cfg, sqlite3 *sql, co rtn = "AND Retain != 0"; break; } - snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE ConnID = '%s' AND " - "PubUser LIKE '%s' %s;", str, connid, user, rtn); + psStmt = sqlite3_mprintf("DELETE FROM %s WHERE ConnID = '%q' AND " + "PubUser LIKE '%q' %s;", str, connid, user, 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 { @@ -353,7 +416,7 @@ mqtt_rtlm_delete_topic(cfg_root_t *cfg, sqlite3 *sql, 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) @@ -375,14 +438,16 @@ mqtt_rtlm_delete_topic(cfg_root_t *cfg, sqlite3 *sql, rtn = "AND Retain != 0"; break; } - snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE ConnID = '%s' AND MsgID = %d AND " - "Topic LIKE '%s' AND PubUser LIKE '%s' AND PubHost LIKE '%s' %s;", str, + 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 { @@ -401,20 +466,20 @@ 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; - 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; @@ -436,19 +501,21 @@ mqtt_rtlm_read_topic(cfg_root_t *cfg, sqlite3 *sql, co mqtt_rtlm_log("Error:: not found topics table name"); return NULL; } - snprintf(szStmt, sizeof szStmt, "SELECT Retain, Topic, Value FROM %s WHERE " - "ConnID = '%s' AND MsgID = %d AND Topic LIKE '%s' %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, 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 = 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 @@ -458,10 +525,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); - s[j].sub_value.msg_base = (u_char*) io_strdup((char*) sqlite3_column_text(stmt, 2)); - s[j].sub_value.msg_len = strlen((char*) s[j].sub_value.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); @@ -487,7 +558,7 @@ mqtt_rtlm_write_subscribe(cfg_root_t *cfg, sqlite3 *sq const char *topic, const char *user, const char *host, char qos) { int ret = 0; - char *str, szStmt[BUFSIZ] = { 0 }; + char *str, *psStmt; sqlite3_stmt *stmt; if (!cfg || !sql || !topic) @@ -498,15 +569,17 @@ mqtt_rtlm_write_subscribe(cfg_root_t *cfg, sqlite3 *sq mqtt_rtlm_log("Error:: not found subscribes table name"); return -1; } - snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (ConnID, MsgID, QoS, Topic, PubUser, " - "PubDate, PubHost) VALUES ('%s', %d, %d, '%s', '%s', " - "datetime('now', 'localtime'), '%s');", str, + 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, 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 { @@ -535,7 +608,7 @@ mqtt_rtlm_delete_subscribe(cfg_root_t *cfg, sqlite3 *s const char *topic, const char *user, const char *host) { int ret = 0; - char *str, szStmt[BUFSIZ] = { 0 }; + char *str, *psStmt; sqlite3_stmt *stmt; if (!cfg || !sql || !topic) @@ -546,14 +619,16 @@ mqtt_rtlm_delete_subscribe(cfg_root_t *cfg, sqlite3 *s mqtt_rtlm_log("Error:: not found subscribes table name"); return -1; } - snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE ConnID = '%s' AND " - "Topic LIKE '%s' AND PubUser LIKE '%s' AND PubHost LIKE '%s';", str, + 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, 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 { @@ -579,7 +654,7 @@ mqtt_subscr_t * mqtt_rtlm_read_subscribe(cfg_root_t *cfg, sqlite3 *sql, const char *connid, const char *topic) { int rowz = 0; - char *str, szStmt[BUFSIZ] = { 0 }; + char *str, *psStmt; sqlite3_stmt *stmt; register int j; mqtt_subscr_t *s = NULL; @@ -592,18 +667,20 @@ mqtt_rtlm_read_subscribe(cfg_root_t *cfg, sqlite3 *sql mqtt_rtlm_log("Error:: not found subscribes table name"); return NULL; } - snprintf(szStmt, sizeof szStmt, "SELECT QoS, Topic FROM %s WHERE ConnID = '%s' AND " - "Topic LIKE '%s';", str, connid, topic); + psStmt = sqlite3_mprintf("SELECT QoS, Topic FROM %s WHERE ConnID = '%q' AND " + "Topic LIKE '%q';", str, connid, topic); - 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 = 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 @@ -613,7 +690,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;