[BACK]Return to pubmqtt.c CVS log [TXT] [DIR] Up to [ELWIX - Embedded LightWeight unIX -] / mqtt / src

Diff for /mqtt/src/pubmqtt.c between versions 1.1.2.10 and 1.2

version 1.1.2.10, 2011/11/28 13:19:39 version 1.2, 2012/01/27 15:05:38
Line 1 Line 1
 #include "global.h"  #include "global.h"
   
   
   extern const char sql_schema[];
   
   
 /*  /*
  * mqtt_db_log() Log database connection message   * mqtt_db_log() Log database connection message
  *   *
Line 36  mqtt_rtlm_open(sl_config *cfg) Line 39  mqtt_rtlm_open(sl_config *cfg)
         if (!cfg)          if (!cfg)
                 return NULL;                  return NULL;
   
           /*
           if (!sqlite3_threadsafe() || sqlite3_config(SQLITE_CONFIG_SERIALIZED))
                   return NULL;
                   */
   
         str = (const char*) cfg_GetAttribute(cfg, CFG("mqtt_pub"), CFG("name"));          str = (const char*) cfg_GetAttribute(cfg, CFG("mqtt_pub"), CFG("name"));
         if (!str) {          if (!str) {
                 mqtt_rtlm_log("Error:: Unknown database name ...\n");                  mqtt_rtlm_log("Error:: Unknown database name ...\n");
                 return NULL;                  return NULL;
         }          }
   
        if (sqlite3_open_v2(str, &sql, SQLITE_OPEN_READWRITE, NULL)) {        if (sqlite3_open_v2(str, &sql, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE, NULL)) {
                 MQTT_RTLM_LOG(sql);                  MQTT_RTLM_LOG(sql);
                 sqlite3_close(sql);                  sqlite3_close(sql);
                 return NULL;                  return NULL;
         }          }
   
           if (sqlite3_exec(sql, sql_schema, NULL, NULL, NULL)) {
                   MQTT_RTLM_LOG(sql);
                   sqlite3_close(sql);
                   return NULL;
           }
         return sql;          return sql;
 }  }
   
Line 68  mqtt_rtlm_close(sqlite3 *sql) Line 81  mqtt_rtlm_close(sqlite3 *sql)
  *   *
  * @cfg = loaded config   * @cfg = loaded config
  * @sql = SQL handle   * @sql = SQL handle
    * @connid = connection id
  * @user = username   * @user = username
  * @host = hostname   * @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   * return: -1 error, 0 session already appears or >0 row changed
  */   */
 int  int
mqtt_rtlm_init_session(sl_config *cfg, sqlite3 *sql, const char *user, const char *host, u_short port)mqtt_rtlm_init_session(sl_config *cfg, sqlite3 *sql, const char *connid, const char *user, 
                 const char *host, char will, ...)
 {  {
           va_list lst;
         int ret = 0;          int ret = 0;
         char *str, szStmt[BUFSIZ] = { 0 };          char *str, szStmt[BUFSIZ] = { 0 };
         sqlite3_stmt *stmt;          sqlite3_stmt *stmt;
Line 88  mqtt_rtlm_init_session(sl_config *cfg, sqlite3 *sql, c Line 105  mqtt_rtlm_init_session(sl_config *cfg, sqlite3 *sql, c
                 mqtt_rtlm_log("Error:: not found online table name");                  mqtt_rtlm_log("Error:: not found online table name");
                 return -1;                  return -1;
         }          }
        snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (Username, RemoteHost, RemotePort) "        if (!will)
                       "VALUES ('%s', '%s', %d);", str, user, host, port);                snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (ConnID, Username, RemoteHost, "
                                 "WillFlag) VALUES ('%s', '%s', '%s', 0);", str, connid, user, host);
         else {
                 va_start(lst, will);
                 snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (ConnID, Username, RemoteHost, "
                                 "WillFlag, WillRetain, WillQoS, WillMsg, WillTopic) "
                                 "VALUES ('%s', '%s', '%s', %d, %d, %d, '%s', '%s');", 
                                 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, szStmt, strlen(szStmt), &stmt, NULL)) {
                 MQTT_RTLM_LOG(sql);                  MQTT_RTLM_LOG(sql);
Line 112  mqtt_rtlm_init_session(sl_config *cfg, sqlite3 *sql, c Line 139  mqtt_rtlm_init_session(sl_config *cfg, sqlite3 *sql, c
  *   *
  * @cfg = loaded config   * @cfg = loaded config
  * @sql = SQL handle   * @sql = SQL handle
    * @connid = connection id
  * @user = username   * @user = username
  * @host = hostname   * @host = hostname
  * return: -1 error, 0 session already appears or >0 row changed   * return: -1 error, 0 session already appears or >0 row changed
  */   */
 int  int
mqtt_rtlm_fini_session(sl_config *cfg, sqlite3 *sql, const char *user, const char *host)mqtt_rtlm_fini_session(sl_config *cfg, sqlite3 *sql, const char *connid, const char *user, const char *host)
 {  {
         int ret = 0;          int ret = 0;
         char *str, szStmt[BUFSIZ] = { 0 };          char *str, szStmt[BUFSIZ] = { 0 };
Line 131  mqtt_rtlm_fini_session(sl_config *cfg, sqlite3 *sql, c Line 159  mqtt_rtlm_fini_session(sl_config *cfg, sqlite3 *sql, c
                 mqtt_rtlm_log("Error:: not found online table name");                  mqtt_rtlm_log("Error:: not found online table name");
                 return -1;                  return -1;
         }          }
        snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE Username = '%s' AND RemoteHost LIKE '%s';",         snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE ConnID = '%s' AND Username = '%s' "
                        str, user, host);                        "AND RemoteHost LIKE '%s';", str, connid, user, host);
   
         if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) {          if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) {
                 MQTT_RTLM_LOG(sql);                  MQTT_RTLM_LOG(sql);
Line 155  mqtt_rtlm_fini_session(sl_config *cfg, sqlite3 *sql, c Line 183  mqtt_rtlm_fini_session(sl_config *cfg, sqlite3 *sql, c
  *   *
  * @cfg = loaded config   * @cfg = loaded config
  * @sql = SQL handle   * @sql = SQL handle
    * @connid = connection id
  * @user = username   * @user = username
  * @host = hostname   * @host = hostname
  * return: -1 error, 0 not logged or >0 logged found rows   * return: -1 error, 0 not logged or >0 logged found rows
  */   */
 int  int
mqtt_rtlm_chk_session(sl_config *cfg, sqlite3 *sql, const char *user, const char *host)mqtt_rtlm_chk_session(sl_config *cfg, sqlite3 *sql, const char *connid, const char *user, const char *host)
 {  {
         int ret = 0;          int ret = 0;
         char *str, szStmt[BUFSIZ] = { 0 };          char *str, szStmt[BUFSIZ] = { 0 };
Line 174  mqtt_rtlm_chk_session(sl_config *cfg, sqlite3 *sql, co Line 203  mqtt_rtlm_chk_session(sl_config *cfg, sqlite3 *sql, co
                 mqtt_rtlm_log("Error:: not found online table name");                  mqtt_rtlm_log("Error:: not found online table name");
                 return -1;                  return -1;
         }          }
        snprintf(szStmt, sizeof szStmt, "SELECT RemoteHost, RemotePort FROM %s WHERE "        snprintf(szStmt, sizeof szStmt, "SELECT ConnID, RemoteHost FROM %s WHERE "
                        "Username = '%s' AND RemoteHost LIKE '%s';", str, user, host);                        "ConnID = '%s' AND Username = '%s' AND RemoteHost LIKE '%s';", 
                         str, connid, user, host);
   
         if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) {          if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) {
                 MQTT_RTLM_LOG(sql);                  MQTT_RTLM_LOG(sql);
Line 195  mqtt_rtlm_chk_session(sl_config *cfg, sqlite3 *sql, co Line 225  mqtt_rtlm_chk_session(sl_config *cfg, sqlite3 *sql, co
  *   *
  * @cfg = loaded config   * @cfg = loaded config
  * @sql = SQL handle   * @sql = SQL handle
    * @msgid = MessageID
  * @topic = topic   * @topic = topic
  * @txt = text   * @txt = text
  * @user = username   * @user = username
Line 203  mqtt_rtlm_chk_session(sl_config *cfg, sqlite3 *sql, co Line 234  mqtt_rtlm_chk_session(sl_config *cfg, sqlite3 *sql, co
  * return: -1 error, 0 no publish or >0 published ok   * return: -1 error, 0 no publish or >0 published ok
  */   */
 int  int
mqtt_rtlm_write_topic(sl_config *cfg, sqlite3 *sql, const char *topic, const char *txt, mqtt_rtlm_write_topic(sl_config *cfg, sqlite3 *sql, u_short msgid, const char *topic, const char *txt, 
                 const char *user, const char *host, char retain)                  const char *user, const char *host, char retain)
 {  {
         int ret = 0;          int ret = 0;
Line 218  mqtt_rtlm_write_topic(sl_config *cfg, sqlite3 *sql, co Line 249  mqtt_rtlm_write_topic(sl_config *cfg, sqlite3 *sql, co
                 mqtt_rtlm_log("Error:: not found topics table name");                  mqtt_rtlm_log("Error:: not found topics table name");
                 return -1;                  return -1;
         }          }
        snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (Retain, Topic, Value, PubUser, PubDate, PubHost) "        snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (Retain, MsgID, Topic, Value, PubUser, "
                        "VALUES (%d, '%s', '%s', '%s', datetime('now', 'localtime'), '%s');", str,                         "PubDate, PubHost) VALUES (%d, %d, '%s', '%s', '%s', "
                        retain, topic, txt, user, host);                        "datetime('now', 'localtime'), '%s');", 
                         str, retain, msgid, topic, txt, user, host);
   
         if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) {          if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) {
                 MQTT_RTLM_LOG(sql);                  MQTT_RTLM_LOG(sql);
Line 243  mqtt_rtlm_write_topic(sl_config *cfg, sqlite3 *sql, co Line 275  mqtt_rtlm_write_topic(sl_config *cfg, sqlite3 *sql, co
  *   *
  * @cfg = loaded config   * @cfg = loaded config
  * @sql = SQL handle   * @sql = SQL handle
    * @msgid = MessageID
  * @topic = topic   * @topic = topic
  * @user = username   * @user = username
  * @host = hostname   * @host = hostname
Line 250  mqtt_rtlm_write_topic(sl_config *cfg, sqlite3 *sql, co Line 283  mqtt_rtlm_write_topic(sl_config *cfg, sqlite3 *sql, co
  * return: -1 error, 0 no changes or >0 deleted rows   * return: -1 error, 0 no changes or >0 deleted rows
  */   */
 int  int
mqtt_rtlm_delete_topic(sl_config *cfg, sqlite3 *sql, const char *topic, mqtt_rtlm_delete_topic(sl_config *cfg, sqlite3 *sql, u_short msgid, const char *topic, 
                 const char *user, const char *host, char retain)                  const char *user, const char *host, char retain)
 {  {
         int ret = 0;          int ret = 0;
Line 276  mqtt_rtlm_delete_topic(sl_config *cfg, sqlite3 *sql, c Line 309  mqtt_rtlm_delete_topic(sl_config *cfg, sqlite3 *sql, c
                         rtn = "AND Retain != 0";                          rtn = "AND Retain != 0";
                         break;                          break;
         }          }
        snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE Topic LIKE '%s' AND "        snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE MsgID = %d AND Topic LIKE '%s' AND "
                         "PubUser LIKE '%s' AND PubHost LIKE '%s' %s;", str,                           "PubUser LIKE '%s' AND PubHost LIKE '%s' %s;", str, 
                        topic, user, host, rtn);                        msgid, topic, user, host, rtn);
   
         if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) {          if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) {
                 MQTT_RTLM_LOG(sql);                  MQTT_RTLM_LOG(sql);
Line 301  mqtt_rtlm_delete_topic(sl_config *cfg, sqlite3 *sql, c Line 334  mqtt_rtlm_delete_topic(sl_config *cfg, sqlite3 *sql, c
  *   *
  * @cfg = loaded config   * @cfg = loaded config
  * @sql = SQL handle   * @sql = SQL handle
    * @msgid = MessageID
  * @topic = topic   * @topic = topic
  * @retain = retain 0 get only dynamic, >0 get only retained and -1 no matter   * @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   * return: NULL error or not found and !=NULL allocated subscribe topics
  */   */
 mqtt_subscr_t *  mqtt_subscr_t *
mqtt_rtlm_read_topic(sl_config *cfg, sqlite3 *sql, const char *topic, char retain)mqtt_rtlm_read_topic(sl_config *cfg, sqlite3 *sql, u_short msgid, const char *topic, char retain)
 {  {
         int rowz = 0;          int rowz = 0;
         char *str, szStr[STRSIZ], szStmt[BUFSIZ] = { 0 };          char *str, szStr[STRSIZ], szStmt[BUFSIZ] = { 0 };
Line 334  mqtt_rtlm_read_topic(sl_config *cfg, sqlite3 *sql, con Line 368  mqtt_rtlm_read_topic(sl_config *cfg, sqlite3 *sql, con
                 mqtt_rtlm_log("Error:: not found topics table name");                  mqtt_rtlm_log("Error:: not found topics table name");
                 return NULL;                  return NULL;
         }          }
        snprintf(szStmt, sizeof szStmt, "SELECT Retain, Topic, Value FROM %s WHERE Topic LIKE '%s' %s;",         snprintf(szStmt, sizeof szStmt, "SELECT Retain, Topic, Value FROM %s WHERE "
                        str, topic, szStr);                        "MsgID = %d AND Topic LIKE '%s' %s;", 
                         str, msgid, topic, szStr);
   
         if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) {          if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) {
                 MQTT_RTLM_LOG(sql);                  MQTT_RTLM_LOG(sql);
Line 354  mqtt_rtlm_read_topic(sl_config *cfg, sqlite3 *sql, con Line 389  mqtt_rtlm_read_topic(sl_config *cfg, sqlite3 *sql, con
   
         /* fill with data */          /* fill with data */
         for (j = 0; j < rowz && sqlite3_step(stmt) == SQLITE_ROW; j++) {          for (j = 0; j < rowz && sqlite3_step(stmt) == SQLITE_ROW; j++) {
                s[j].sub_ret = *sqlite3_column_text(stmt, 0);                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._base = (u_char*) strdup((char*) sqlite3_column_text(stmt, 1));
                s[j].sub_topic._size = strlen(s[j].sub_topic._base);                s[j].sub_topic._size = strlen((char*) s[j].sub_topic._base);
                s[j].sub_value._base = strdup(sqlite3_column_text(stmt, 2));                s[j].sub_value._base = (u_char*) strdup((char*) sqlite3_column_text(stmt, 2));
                s[j].sub_value._size = strlen(s[j].sub_value._base);                s[j].sub_value._size = strlen((char*) s[j].sub_value._base);
         }          }
 end:  end:
         sqlite3_finalize(stmt);          sqlite3_finalize(stmt);
Removed from v.1.1.2.10  
changed lines
  Added in v.1.2

FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>