version 1.2.2.1, 2012/01/27 15:18:12
|
version 1.2.2.2, 2012/02/01 15:44:40
|
Line 400 end:
|
Line 400 end:
|
|
|
return s; |
return s; |
} |
} |
|
|
|
/* |
|
* mqtt_rtlm_write_subscribe() Subscribe topic |
|
* |
|
* @cfg = loaded config |
|
* @sql = SQL handle |
|
* @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(sl_config *cfg, sqlite3 *sql, u_short msgid, const char *topic, |
|
const char *user, const char *host, char qos) |
|
{ |
|
int ret = 0; |
|
char *str, szStmt[BUFSIZ] = { 0 }; |
|
sqlite3_stmt *stmt; |
|
|
|
if (!cfg || !sql || !topic) |
|
return -1; |
|
|
|
str = (char*) cfg_GetAttribute(cfg, CFG("mqtt_pub"), CFG("tbl_subscribes")); |
|
if (!str) { |
|
mqtt_rtlm_log("Error:: not found subscribes table name"); |
|
return -1; |
|
} |
|
snprintf(szStmt, sizeof szStmt, "INSERT INTO %s (MsgID, QoS, Topic, PubUser, " |
|
"PubDate, PubHost) VALUES (%d, %d, '%s', '%s', " |
|
"datetime('now', 'localtime'), '%s');", str, |
|
msgid, qos, topic, user, host); |
|
|
|
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_subscribe() Delete subscribe |
|
* |
|
* @cfg = loaded config |
|
* @sql = SQL handle |
|
* @topic = topic |
|
* @user = username |
|
* @host = hostname |
|
* @qos = Subscribe QoS if -1 no matter |
|
* return: -1 error, 0 no changes or >0 deleted rows |
|
*/ |
|
int |
|
mqtt_rtlm_delete_subscribe(sl_config *cfg, sqlite3 *sql, const char *topic, |
|
const char *user, const char *host, char qos) |
|
{ |
|
int ret = 0; |
|
char *str, szStr[STRSIZ] = { 0 }, szStmt[BUFSIZ] = { 0 }; |
|
sqlite3_stmt *stmt; |
|
|
|
if (!cfg || !sql || !topic) |
|
return -1; |
|
|
|
str = (char*) cfg_GetAttribute(cfg, CFG("mqtt_pub"), CFG("tbl_subscribes")); |
|
if (!str) { |
|
mqtt_rtlm_log("Error:: not found subscribes table name"); |
|
return -1; |
|
} |
|
if (qos > -1 && qos < 3) |
|
snprintf(szStr, sizeof szStr, "AND QoS = %d", qos); |
|
snprintf(szStmt, sizeof szStmt, "DELETE FROM %s WHERE Topic LIKE '%s' AND " |
|
"PubUser LIKE '%s' AND PubHost LIKE '%s' %s;", str, |
|
topic, user, host, szStr); |
|
|
|
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_read_subscribe() Get subscribe topic |
|
* |
|
* @cfg = loaded config |
|
* @sql = SQL handle |
|
* @topic = topic |
|
* return: NULL error or not found and !=NULL allocated subscribe topics |
|
*/ |
|
mqtt_subscr_t * |
|
mqtt_rtlm_read_subscribe(sl_config *cfg, sqlite3 *sql, const char *topic) |
|
{ |
|
int rowz = 0; |
|
char *str, szStmt[BUFSIZ] = { 0 }; |
|
sqlite3_stmt *stmt; |
|
register int j; |
|
mqtt_subscr_t *s = NULL; |
|
|
|
if (!cfg || !sql || !topic) |
|
return NULL; |
|
|
|
str = (char*) cfg_GetAttribute(cfg, CFG("mqtt_pub"), CFG("tbl_subscribes")); |
|
if (!str) { |
|
mqtt_rtlm_log("Error:: not found subscribes table name"); |
|
return NULL; |
|
} |
|
snprintf(szStmt, sizeof szStmt, "SELECT QoS, Topic FROM %s WHERE Topic LIKE '%s';", str, topic); |
|
|
|
if (sqlite3_prepare_v2(sql, szStmt, strlen(szStmt), &stmt, NULL)) { |
|
MQTT_RTLM_LOG(sql); |
|
return NULL; |
|
} |
|
|
|
/* 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); |
|
s[j].sub_value.msg_base = NULL; |
|
s[j].sub_value.msg_len = 0; |
|
} |
|
end: |
|
sqlite3_finalize(stmt); |
|
|
|
return s; |
|
} |