[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.2.2.3 and 1.2.2.9

version 1.2.2.3, 2012/04/11 15:08:27 version 1.2.2.9, 2012/05/27 10:04:05
Line 20  mqtt_rtlm_log(const char *fmt, ...) Line 20  mqtt_rtlm_log(const char *fmt, ...)
         vsyslog(LOG_ERR, fmt, lst);          vsyslog(LOG_ERR, fmt, lst);
         va_end(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))))                                          sqlite3_errcode((_sql)), sqlite3_errmsg((_sql))))
   
   
Line 39  mqtt_rtlm_open(cfg_root_t *cfg) Line 40  mqtt_rtlm_open(cfg_root_t *cfg)
         if (!cfg)          if (!cfg)
                 return NULL;                  return NULL;
   
         /*  
         if (!sqlite3_threadsafe() || sqlite3_config(SQLITE_CONFIG_SERIALIZED))  
                 return NULL;  
                 */  
   
         str = cfg_getAttribute(cfg, "mqtt_pub", "name");          str = cfg_getAttribute(cfg, "mqtt_pub", "name");
         if (!str) {          if (!str) {
                 mqtt_rtlm_log("Error:: Unknown database name ...\n");                  mqtt_rtlm_log("Error:: Unknown database name ...\n");
Line 204  mqtt_rtlm_chk_session(cfg_root_t *cfg, sqlite3 *sql, c Line 200  mqtt_rtlm_chk_session(cfg_root_t *cfg, sqlite3 *sql, c
                 return -1;                  return -1;
         }          }
         snprintf(szStmt, sizeof szStmt, "SELECT ConnID, RemoteHost FROM %s WHERE "          snprintf(szStmt, sizeof szStmt, "SELECT ConnID, RemoteHost FROM %s WHERE "
                        "ConnID = '%s' AND Username = '%s' AND RemoteHost LIKE '%s';",                         "ConnID = '%s' AND Username LIKE '%s' AND RemoteHost LIKE '%s';", 
                         str, connid, user, host);                          str, connid, user, host);
   
         if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) {          if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) {
Line 225  mqtt_rtlm_chk_session(cfg_root_t *cfg, sqlite3 *sql, c Line 221  mqtt_rtlm_chk_session(cfg_root_t *cfg, sqlite3 *sql, c
  *   *
  * @cfg = loaded config   * @cfg = loaded config
  * @sql = SQL handle   * @sql = SQL handle
    * @connid = connection id
  * @msgid = MessageID   * @msgid = MessageID
  * @topic = topic   * @topic = topic
  * @txt = text   * @txt = text
Line 234  mqtt_rtlm_chk_session(cfg_root_t *cfg, sqlite3 *sql, c Line 231  mqtt_rtlm_chk_session(cfg_root_t *cfg, sqlite3 *sql, c
  * 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(cfg_root_t *cfg, sqlite3 *sql, u_short msgid, const char *topic, const char *txt, mqtt_rtlm_write_topic(cfg_root_t *cfg, sqlite3 *sql, const char *connid, u_short msgid, 
                const char *user, const char *host, char retain)                const char *topic, const char *txt, const char *user, const char *host, char retain)
 {  {
         int ret = 0;          int ret = 0;
         char *str, szStmt[BUFSIZ] = { 0 };          char *str, szStmt[BUFSIZ] = { 0 };
Line 249  mqtt_rtlm_write_topic(cfg_root_t *cfg, sqlite3 *sql, u Line 246  mqtt_rtlm_write_topic(cfg_root_t *cfg, sqlite3 *sql, u
                 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, MsgID, Topic, Value, PubUser, "        snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (Retain, ConnID, MsgID, Topic, Value, PubUser, "
                        "PubDate, PubHost) VALUES (%d, %d, '%s', '%s', '%s', "                        "PubDate, PubHost) VALUES (%d, '%s', %d, '%s', '%s', '%s', "
                         "datetime('now', 'localtime'), '%s');",                           "datetime('now', 'localtime'), '%s');", 
                        str, retain, msgid, topic, txt, user, host);                        str, retain, connid, 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 271  mqtt_rtlm_write_topic(cfg_root_t *cfg, sqlite3 *sql, u Line 268  mqtt_rtlm_write_topic(cfg_root_t *cfg, sqlite3 *sql, u
 }  }
   
 /*  /*
    * 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, szStmt[BUFSIZ] = { 0 };
           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;
           }
           snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE ConnID = '%s' AND "
                           "PubUser LIKE '%s' %s;", str, connid, user, rtn);
   
           if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) {
                   MQTT_RTLM_LOG(sql);
                   return -1;
           }
           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   * mqtt_rtlm_delete_topic() Delete topic
  *   *
  * @cfg = loaded config   * @cfg = loaded config
  * @sql = SQL handle   * @sql = SQL handle
    * @connid = connection id
  * @msgid = MessageID   * @msgid = MessageID
  * @topic = topic   * @topic = topic
  * @user = username   * @user = username
Line 283  mqtt_rtlm_write_topic(cfg_root_t *cfg, sqlite3 *sql, u Line 336  mqtt_rtlm_write_topic(cfg_root_t *cfg, sqlite3 *sql, u
  * 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(cfg_root_t *cfg, sqlite3 *sql, u_short msgid, const char *topic, mqtt_rtlm_delete_topic(cfg_root_t *cfg, sqlite3 *sql, const char *connid, u_short msgid, 
                const char *user, const char *host, char retain)                const char *topic, const char *user, const char *host, char retain)
 {  {
         int ret = 0;          int ret = 0;
         char *str, *rtn, szStmt[BUFSIZ] = { 0 };          char *str, *rtn, szStmt[BUFSIZ] = { 0 };
Line 309  mqtt_rtlm_delete_topic(cfg_root_t *cfg, sqlite3 *sql,  Line 362  mqtt_rtlm_delete_topic(cfg_root_t *cfg, sqlite3 *sql, 
                         rtn = "AND Retain != 0";                          rtn = "AND Retain != 0";
                         break;                          break;
         }          }
        snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE MsgID = %d AND Topic LIKE '%s' AND "        snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE ConnID = '%s' AND MsgID = %d AND "
                        "PubUser LIKE '%s' AND PubHost LIKE '%s' %s;", str,                         "Topic LIKE '%s' AND PubUser LIKE '%s' AND PubHost LIKE '%s' %s;", str, 
                        msgid, topic, user, host, rtn);                        connid, 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 334  mqtt_rtlm_delete_topic(cfg_root_t *cfg, sqlite3 *sql,  Line 387  mqtt_rtlm_delete_topic(cfg_root_t *cfg, sqlite3 *sql, 
  *   *
  * @cfg = loaded config   * @cfg = loaded config
  * @sql = SQL handle   * @sql = SQL handle
    * @connid = connection id
  * @msgid = MessageID   * @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(cfg_root_t *cfg, sqlite3 *sql, u_short msgid, const char *topic, char retain)mqtt_rtlm_read_topic(cfg_root_t *cfg, sqlite3 *sql, const char *connid, 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 369  mqtt_rtlm_read_topic(cfg_root_t *cfg, sqlite3 *sql, u_ Line 424  mqtt_rtlm_read_topic(cfg_root_t *cfg, sqlite3 *sql, u_
                 return NULL;                  return NULL;
         }          }
         snprintf(szStmt, sizeof szStmt, "SELECT Retain, Topic, Value FROM %s WHERE "          snprintf(szStmt, sizeof szStmt, "SELECT Retain, Topic, Value FROM %s WHERE "
                        "MsgID = %d AND Topic LIKE '%s' %s;",                         "ConnID = '%s' AND MsgID = %d AND Topic LIKE '%s' %s;", 
                        str, msgid, topic, szStr);                        str, connid, 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 380  mqtt_rtlm_read_topic(cfg_root_t *cfg, sqlite3 *sql, u_ Line 435  mqtt_rtlm_read_topic(cfg_root_t *cfg, sqlite3 *sql, u_
         /* calculate count of rows and allocate subscribe items */          /* calculate count of rows and allocate subscribe items */
         while (sqlite3_step(stmt) == SQLITE_ROW)          while (sqlite3_step(stmt) == SQLITE_ROW)
                 rowz++;                  rowz++;
        if (!(s = malloc((rowz + 1) * sizeof(mqtt_subscr_t)))) {        if (!(s = io_malloc((rowz + 1) * sizeof(mqtt_subscr_t)))) {
                 mqtt_rtlm_log("Error:: System #%d - %s", errno, strerror(errno));                  mqtt_rtlm_log("Error:: System #%d - %s", errno, strerror(errno));
                 goto end;                  goto end;
         } else          } else
Line 390  mqtt_rtlm_read_topic(cfg_root_t *cfg, sqlite3 *sql, u_ Line 445  mqtt_rtlm_read_topic(cfg_root_t *cfg, sqlite3 *sql, u_
         /* 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 = (char) sqlite3_column_int(stmt, 0);                  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_base = (u_char*) io_strdup((char*) sqlite3_column_text(stmt, 1));
                 s[j].sub_topic.msg_len = strlen((char*) s[j].sub_topic.msg_base);                  s[j].sub_topic.msg_len = strlen((char*) s[j].sub_topic.msg_base);
                s[j].sub_value.msg_base = (u_char*) strdup((char*) sqlite3_column_text(stmt, 2));                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);                  s[j].sub_value.msg_len = strlen((char*) s[j].sub_value.msg_base);
         }          }
 end:  end:
Line 406  end: Line 461  end:
  *   *
  * @cfg = loaded config   * @cfg = loaded config
  * @sql = SQL handle   * @sql = SQL handle
    * @connid = connection id
  * @msgid = MessageID   * @msgid = MessageID
  * @topic = topic   * @topic = topic
  * @user = username   * @user = username
Line 414  end: Line 470  end:
  * 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_subscribe(cfg_root_t *cfg, sqlite3 *sql, u_short msgid, const char *topic, mqtt_rtlm_write_subscribe(cfg_root_t *cfg, sqlite3 *sql, const char *connid, u_short msgid, 
                const char *user, const char *host, char qos)                const char *topic, const char *user, const char *host, char qos)
 {  {
         int ret = 0;          int ret = 0;
         char *str, szStmt[BUFSIZ] = { 0 };          char *str, szStmt[BUFSIZ] = { 0 };
Line 429  mqtt_rtlm_write_subscribe(cfg_root_t *cfg, sqlite3 *sq Line 485  mqtt_rtlm_write_subscribe(cfg_root_t *cfg, sqlite3 *sq
                 mqtt_rtlm_log("Error:: not found subscribes table name");                  mqtt_rtlm_log("Error:: not found subscribes table name");
                 return -1;                  return -1;
         }          }
        snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (MsgID, QoS, Topic, PubUser, "        snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (ConnID, MsgID, QoS, Topic, PubUser, "
                        "PubDate, PubHost) VALUES (%d, %d, '%s', '%s', "                        "PubDate, PubHost) VALUES ('%s', %d, %d, '%s', '%s', "
                         "datetime('now', 'localtime'), '%s');", str,                           "datetime('now', 'localtime'), '%s');", str, 
                        msgid, qos, topic, user, host);                        connid, msgid, qos, topic, 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 455  mqtt_rtlm_write_subscribe(cfg_root_t *cfg, sqlite3 *sq Line 511  mqtt_rtlm_write_subscribe(cfg_root_t *cfg, sqlite3 *sq
  *   *
  * @cfg = loaded config   * @cfg = loaded config
  * @sql = SQL handle   * @sql = SQL handle
    * @connid = connection id
  * @topic = topic   * @topic = topic
  * @user = username   * @user = username
  * @host = hostname   * @host = hostname
  * @qos = Subscribe QoS if -1 no matter  
  * 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_subscribe(cfg_root_t *cfg, sqlite3 *sql, const char *topic, mqtt_rtlm_delete_subscribe(cfg_root_t *cfg, sqlite3 *sql, const char *connid, 
                const char *user, const char *host, char qos)                const char *topic, const char *user, const char *host)
 {  {
         int ret = 0;          int ret = 0;
        char *str, szStr[STRSIZ] = { 0 }, szStmt[BUFSIZ] = { 0 };        char *str, szStmt[BUFSIZ] = { 0 };
         sqlite3_stmt *stmt;          sqlite3_stmt *stmt;
   
         if (!cfg || !sql || !topic)          if (!cfg || !sql || !topic)
Line 477  mqtt_rtlm_delete_subscribe(cfg_root_t *cfg, sqlite3 *s Line 533  mqtt_rtlm_delete_subscribe(cfg_root_t *cfg, sqlite3 *s
                 mqtt_rtlm_log("Error:: not found subscribes table name");                  mqtt_rtlm_log("Error:: not found subscribes table name");
                 return -1;                  return -1;
         }          }
        if (qos > -1 && qos < 3)        snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE ConnID = '%s' AND "
                snprintf(szStr, sizeof szStr, "AND QoS = %d", qos);                        "Topic LIKE '%s' AND PubUser LIKE '%s' AND PubHost LIKE '%s';", str, 
        snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE Topic LIKE '%s' AND "                        connid, topic, user, host);
                        "PubUser LIKE '%s' AND PubHost LIKE '%s' %s;", str,  
                        topic, user, host, 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 504  mqtt_rtlm_delete_subscribe(cfg_root_t *cfg, sqlite3 *s Line 558  mqtt_rtlm_delete_subscribe(cfg_root_t *cfg, sqlite3 *s
  *   *
  * @cfg = loaded config   * @cfg = loaded config
  * @sql = SQL handle   * @sql = SQL handle
    * @connid = connection id
  * @topic = topic   * @topic = topic
  * 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_subscribe(cfg_root_t *cfg, sqlite3 *sql, const char *topic)mqtt_rtlm_read_subscribe(cfg_root_t *cfg, sqlite3 *sql, const char *connid, const char *topic)
 {  {
         int rowz = 0;          int rowz = 0;
         char *str, szStmt[BUFSIZ] = { 0 };          char *str, szStmt[BUFSIZ] = { 0 };
Line 524  mqtt_rtlm_read_subscribe(cfg_root_t *cfg, sqlite3 *sql Line 579  mqtt_rtlm_read_subscribe(cfg_root_t *cfg, sqlite3 *sql
                 mqtt_rtlm_log("Error:: not found subscribes table name");                  mqtt_rtlm_log("Error:: not found subscribes table name");
                 return NULL;                  return NULL;
         }          }
        snprintf(szStmt, sizeof szStmt, "SELECT QoS, Topic FROM %s WHERE Topic LIKE '%s';", str, topic);        snprintf(szStmt, sizeof szStmt, "SELECT QoS, Topic FROM %s WHERE ConnID = '%s' AND "
                         "Topic LIKE '%s';", str, connid, topic);
   
         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 534  mqtt_rtlm_read_subscribe(cfg_root_t *cfg, sqlite3 *sql Line 590  mqtt_rtlm_read_subscribe(cfg_root_t *cfg, sqlite3 *sql
         /* calculate count of rows and allocate subscribe items */          /* calculate count of rows and allocate subscribe items */
         while (sqlite3_step(stmt) == SQLITE_ROW)          while (sqlite3_step(stmt) == SQLITE_ROW)
                 rowz++;                  rowz++;
        if (!(s = malloc((rowz + 1) * sizeof(mqtt_subscr_t)))) {        if (!(s = io_malloc((rowz + 1) * sizeof(mqtt_subscr_t)))) {
                 mqtt_rtlm_log("Error:: System #%d - %s", errno, strerror(errno));                  mqtt_rtlm_log("Error:: System #%d - %s", errno, strerror(errno));
                 goto end;                  goto end;
         } else          } else
Line 544  mqtt_rtlm_read_subscribe(cfg_root_t *cfg, sqlite3 *sql Line 600  mqtt_rtlm_read_subscribe(cfg_root_t *cfg, sqlite3 *sql
         /* 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 = (char) sqlite3_column_int(stmt, 0);                  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_base = (u_char*) io_strdup((char*) sqlite3_column_text(stmt, 1));
                 s[j].sub_topic.msg_len = strlen((char*) s[j].sub_topic.msg_base);                  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_base = NULL;
                 s[j].sub_value.msg_len = 0;                  s[j].sub_value.msg_len = 0;
Removed from v.1.2.2.3  
changed lines
  Added in v.1.2.2.9

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