Sqlite persistence plugin.

This commit is contained in:
Roger A. Light
2022-01-09 23:54:09 +00:00
parent 0a0f4ead42
commit dc906c5221
35 changed files with 3459 additions and 2 deletions
+3
View File
@@ -132,6 +132,9 @@ WITH_OLD_KEEPALIVE=no
# broker plugins from working.
WITH_LTO=no
# Build with sqlite3 support - this enables the sqlite persistence plugin.
WITH_SQLITE=yes
# =============================================================================
# End of user configuration
# =============================================================================
+2 -1
View File
@@ -1,6 +1,7 @@
DIRS= \
dynamic-security \
examples
examples \
persist-sqlite
.PHONY : all binary check clean reallyclean test install uninstall
+45
View File
@@ -0,0 +1,45 @@
if(SQLITE3_FOUND)
set(CLIENT_INC
"${SQLITE3_INCLUDE_DIRS}"
"${PTHREAD_INCLUDE_DIR}"
"${STDBOOL_H_PATH}"
"${STDINT_H_PATH}"
"${mosquitto_SOURCE_DIR}"
"${mosquitto_SOURCE_DIR}/deps"
"${mosquitto_SOURCE_DIR}/include"
"${mosquitto_SOURCE_DIR}/src"
)
set(CLIENT_DIR "${mosquitto_BINARY_DIR}/lib" "${SQLITE3_DIR}")
add_library(mosquitto_persist_sqlite MODULE
clients.c
client_msgs.c
init.c
msg_store.c
plugin.c
restore.c
retains.c
subscriptions.c
)
target_include_directories(mosquitto_persist_sqlite PRIVATE
${CLIENT_INC} ${SQLITE3_INCLUDE_DIR}
)
link_directories(${CLIENT_DIR} "${mosquitto_SOURCE_DIR}")
set_target_properties(mosquitto_persist_sqlite PROPERTIES
PREFIX ""
POSITION_INDEPENDENT_CODE 1
)
target_link_libraries(mosquitto_persist_sqlite ${SQLITE3_LIBRARIES})
if(WIN32)
target_link_libraries(mosquitto_persist_sqlite mosquitto)
endif()
install(TARGETS mosquitto_persist_sqlite
RUNTIME DESTINATION "${CMAKE_INSTALL_BINDIR}"
LIBRARY DESTINATION "${CMAKE_INSTALL_LIBDIR}"
)
endif()
+68
View File
@@ -0,0 +1,68 @@
include ../../config.mk
.PHONY : all binary check clean reallyclean test install uninstall
PLUGIN_NAME=mosquitto_persist_sqlite
LOCAL_CPPFLAGS=-I../../src/
OBJS= \
clients.o \
client_msgs.o \
init.o \
msg_store.o \
plugin.o \
restore.o \
retains.o \
subscriptions.o
ifeq ($(WITH_SQLITE),yes)
ALL_DEPS:= binary
else
ALL_DEPS:=
endif
all : ${ALL_DEPS}
binary : ${PLUGIN_NAME}.so
${PLUGIN_NAME}.so : ${OBJS}
${CROSS_COMPILE}${CC} $(PLUGIN_LDFLAGS) -fPIC -shared $^ -o $@ -lsqlite3
clients.o : clients.c persist_sqlite.h
${CROSS_COMPILE}${CC} $(LOCAL_CPPFLAGS) $(PLUGIN_CPPFLAGS) $(PLUGIN_CFLAGS) -c $< -o $@
client_msgs.o : client_msgs.c persist_sqlite.h
${CROSS_COMPILE}${CC} $(LOCAL_CPPFLAGS) $(PLUGIN_CPPFLAGS) $(PLUGIN_CFLAGS) -c $< -o $@
init.o : init.c persist_sqlite.h
${CROSS_COMPILE}${CC} $(LOCAL_CPPFLAGS) $(PLUGIN_CPPFLAGS) $(PLUGIN_CFLAGS) -c $< -o $@
msg_store.o : msg_store.c persist_sqlite.h
${CROSS_COMPILE}${CC} $(LOCAL_CPPFLAGS) $(PLUGIN_CPPFLAGS) $(PLUGIN_CFLAGS) -c $< -o $@
plugin.o : plugin.c persist_sqlite.h
${CROSS_COMPILE}${CC} $(LOCAL_CPPFLAGS) $(PLUGIN_CPPFLAGS) $(PLUGIN_CFLAGS) -c $< -o $@
restore.o : restore.c persist_sqlite.h
${CROSS_COMPILE}${CC} $(LOCAL_CPPFLAGS) $(PLUGIN_CPPFLAGS) $(PLUGIN_CFLAGS) -c $< -o $@
retains.o : retains.c persist_sqlite.h
${CROSS_COMPILE}${CC} $(LOCAL_CPPFLAGS) $(PLUGIN_CPPFLAGS) $(PLUGIN_CFLAGS) -c $< -o $@
subscriptions.o : subscriptions.c persist_sqlite.h
${CROSS_COMPILE}${CC} $(LOCAL_CPPFLAGS) $(PLUGIN_CPPFLAGS) $(PLUGIN_CFLAGS) -c $< -o $@
reallyclean : clean
clean:
-rm -f *.o ${PLUGIN_NAME}.so *.gcda *.gcno
check: test
test:
install: all
ifeq ($(WITH_SQLITE),yes)
$(INSTALL) -d "${DESTDIR}$(libdir)"
$(INSTALL) ${STRIP_OPTS} ${PLUGIN_NAME}.so "${DESTDIR}${libdir}/${PLUGIN_NAME}.so"
endif
uninstall :
-rm -f "${DESTDIR}${libdir}/${PLUGIN_NAME}.so"
+133
View File
@@ -0,0 +1,133 @@
/*
Copyright (c) 2021 Roger Light <roger@atchoo.org>
All rights reserved. This program and the accompanying materials
are made available under the terms of the Eclipse Public License 2.0
and Eclipse Distribution License v1.0 which accompany this distribution.
The Eclipse Public License is available at
https://www.eclipse.org/legal/epl-2.0/
and the Eclipse Distribution License is available at
http://www.eclipse.org/org/documents/edl-v10.php.
SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
Contributors:
Roger Light - initial implementation and documentation.
*/
#include <string.h>
#include <sqlite3.h>
#include "mosquitto.h"
#include "mosquitto_broker.h"
#include "persist_sqlite.h"
int persist_sqlite__client_msg_add_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_evt_persist_client_msg *ed = event_data;
struct mosquitto_sqlite *ms = userdata;
int rc = MOSQ_ERR_UNKNOWN;
UNUSED(event);
if(sqlite3_bind_text(ms->client_msg_add_stmt, 1, ed->client_id, (int)strlen(ed->client_id), SQLITE_STATIC) == SQLITE_OK
&& sqlite3_bind_int64(ms->client_msg_add_stmt, 2, (int64_t)ed->store_id) == SQLITE_OK
&& sqlite3_bind_int(ms->client_msg_add_stmt, 3, ed->dup) == SQLITE_OK
&& sqlite3_bind_int(ms->client_msg_add_stmt, 4, ed->direction) == SQLITE_OK
&& sqlite3_bind_int(ms->client_msg_add_stmt, 5, ed->mid) == SQLITE_OK
&& sqlite3_bind_int(ms->client_msg_add_stmt, 6, ed->qos) == SQLITE_OK
&& sqlite3_bind_int(ms->client_msg_add_stmt, 7, ed->retain) == SQLITE_OK
&& sqlite3_bind_int(ms->client_msg_add_stmt, 8, ed->state) == SQLITE_OK
){
rc = sqlite3_step(ms->client_msg_add_stmt);
if(rc == SQLITE_DONE){
rc = MOSQ_ERR_SUCCESS;
}else{
rc = MOSQ_ERR_UNKNOWN;
}
}
sqlite3_reset(ms->client_msg_add_stmt);
return rc;
}
int persist_sqlite__client_msg_remove_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_evt_persist_client_msg *ed = event_data;
struct mosquitto_sqlite *ms = userdata;
int rc = 1;
UNUSED(event);
if(sqlite3_bind_text(ms->client_msg_remove_stmt, 1, ed->client_id, (int)strlen(ed->client_id), SQLITE_STATIC) == SQLITE_OK
&& sqlite3_bind_int64(ms->client_msg_remove_stmt, 2, (int64_t)ed->store_id) == SQLITE_OK
&& sqlite3_bind_int(ms->client_msg_remove_stmt, 3, ed->direction) == SQLITE_OK
){
rc = sqlite3_step(ms->client_msg_remove_stmt);
if(rc == SQLITE_DONE){
rc = MOSQ_ERR_SUCCESS;
}else{
rc = MOSQ_ERR_UNKNOWN;
}
}
sqlite3_reset(ms->client_msg_remove_stmt);
return rc;
}
int persist_sqlite__client_msg_update_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_evt_persist_client_msg *ed = event_data;
struct mosquitto_sqlite *ms = userdata;
int rc = MOSQ_ERR_UNKNOWN;
UNUSED(event);
if(sqlite3_bind_int(ms->client_msg_update_stmt, 1, ed->state) == SQLITE_OK
&& sqlite3_bind_int(ms->client_msg_update_stmt, 2, ed->dup) == SQLITE_OK
&& sqlite3_bind_text(ms->client_msg_update_stmt, 3, ed->client_id, (int)strlen(ed->client_id), SQLITE_STATIC) == SQLITE_OK
&& sqlite3_bind_int64(ms->client_msg_update_stmt, 4, (int64_t)ed->store_id) == SQLITE_OK
){
rc = sqlite3_step(ms->client_msg_update_stmt);
if(rc == SQLITE_DONE){
rc = MOSQ_ERR_SUCCESS;
}else{
rc = MOSQ_ERR_UNKNOWN;
}
}
sqlite3_reset(ms->client_msg_update_stmt);
return rc;
}
int persist_sqlite__client_msg_clear_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_evt_persist_client_msg *ed = event_data;
struct mosquitto_sqlite *ms = userdata;
int rc = 1;
UNUSED(event);
if(sqlite3_bind_text(ms->client_msg_clear_stmt, 1, ed->client_id, (int)strlen(ed->client_id), SQLITE_STATIC) == SQLITE_OK
&& sqlite3_bind_int64(ms->client_msg_clear_stmt, 2, ed->direction) == SQLITE_OK
){
rc = sqlite3_step(ms->client_msg_clear_stmt);
if(rc == SQLITE_DONE){
rc = MOSQ_ERR_SUCCESS;
}else{
rc = MOSQ_ERR_UNKNOWN;
}
}
sqlite3_reset(ms->client_msg_clear_stmt);
return rc;
}
+122
View File
@@ -0,0 +1,122 @@
/*
Copyright (c) 2021 Roger Light <roger@atchoo.org>
All rights reserved. This program and the accompanying materials
are made available under the terms of the Eclipse Public License 2.0
and Eclipse Distribution License v1.0 which accompany this distribution.
The Eclipse Public License is available at
https://www.eclipse.org/legal/epl-2.0/
and the Eclipse Distribution License is available at
http://www.eclipse.org/org/documents/edl-v10.php.
SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
Contributors:
Roger Light - initial implementation and documentation.
*/
#include <string.h>
#include <sqlite3.h>
#include "mosquitto.h"
#include "mosquitto_broker.h"
#include "persist_sqlite.h"
int persist_sqlite__client_add_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_evt_persist_client *ed = event_data;
struct mosquitto_sqlite *ms = userdata;
int rc = MOSQ_ERR_UNKNOWN;
UNUSED(event);
if(sqlite3_bind_text(ms->client_add_stmt, 1,
ed->client_id, (int)strlen(ed->client_id), SQLITE_STATIC) == SQLITE_OK){
if(ed->username){
sqlite3_bind_text(ms->client_add_stmt, 2,
ed->username, (int)strlen(ed->username),
SQLITE_STATIC);
}else{
sqlite3_bind_null(ms->client_add_stmt, 2);
}
if(sqlite3_bind_int64(ms->client_add_stmt, 3, ed->will_delay_time) == SQLITE_OK
&& sqlite3_bind_int64(ms->client_add_stmt, 4, ed->session_expiry_time) == SQLITE_OK
&& sqlite3_bind_int(ms->client_add_stmt, 5, ed->listener_port) == SQLITE_OK
&& sqlite3_bind_int(ms->client_add_stmt, 6, (int)ed->max_packet_size) == SQLITE_OK
&& sqlite3_bind_int(ms->client_add_stmt, 7, ed->max_qos) == SQLITE_OK
&& sqlite3_bind_int(ms->client_add_stmt, 8, ed->retain_available) == SQLITE_OK
&& sqlite3_bind_int(ms->client_add_stmt, 9, (int)ed->session_expiry_interval) == SQLITE_OK
&& sqlite3_bind_int(ms->client_add_stmt, 10, (int)ed->will_delay_interval) == SQLITE_OK
){
rc = sqlite3_step(ms->client_add_stmt);
if(rc == SQLITE_DONE){
rc = MOSQ_ERR_SUCCESS;
}else{
rc = MOSQ_ERR_UNKNOWN;
}
}
}
sqlite3_reset(ms->client_add_stmt);
return rc;
}
int persist_sqlite__client_remove_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_evt_persist_client *ed = event_data;
struct mosquitto_sqlite *ms = userdata;
int rc = 1;
UNUSED(event);
if(sqlite3_bind_text(ms->subscription_clear_stmt, 1,
ed->client_id, (int)strlen(ed->client_id), SQLITE_STATIC) == SQLITE_OK){
sqlite3_step(ms->subscription_clear_stmt);
sqlite3_reset(ms->subscription_clear_stmt);
}
if(sqlite3_bind_text(ms->client_remove_stmt, 1,
ed->client_id, (int)strlen(ed->client_id), SQLITE_STATIC) == SQLITE_OK){
rc = sqlite3_step(ms->client_remove_stmt);
if(rc == SQLITE_DONE){
rc = MOSQ_ERR_SUCCESS;
}else{
rc = MOSQ_ERR_UNKNOWN;
}
}
sqlite3_reset(ms->client_remove_stmt);
return rc;
}
int persist_sqlite__client_update_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_evt_persist_client *ed = event_data;
struct mosquitto_sqlite *ms = userdata;
int rc = 1;
UNUSED(event);
if(sqlite3_bind_int64(ms->client_update_stmt, 1, ed->session_expiry_time) == SQLITE_OK
&& sqlite3_bind_int64(ms->client_update_stmt, 2, ed->will_delay_time) == SQLITE_OK
&& sqlite3_bind_text(ms->client_update_stmt, 3, ed->client_id,
(int)strlen(ed->client_id), SQLITE_STATIC) == SQLITE_OK
){
rc = sqlite3_step(ms->client_update_stmt);
if(rc == SQLITE_DONE){
rc = MOSQ_ERR_SUCCESS;
}else{
rc = MOSQ_ERR_UNKNOWN;
}
}
sqlite3_reset(ms->client_update_stmt);
return rc;
}
+308
View File
@@ -0,0 +1,308 @@
/*
Copyright (c) 2021 Roger Light <roger@atchoo.org>
All rights reserved. This program and the accompanying materials
are made available under the terms of the Eclipse Public License 2.0
and Eclipse Distribution License v1.0 which accompany this distribution.
The Eclipse Public License is available at
https://www.eclipse.org/legal/epl-2.0/
and the Eclipse Distribution License is available at
http://www.eclipse.org/org/documents/edl-v10.php.
SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
Contributors:
Roger Light - initial implementation and documentation.
*/
#include <stdio.h>
#include <sqlite3.h>
#include "persist_sqlite.h"
#include "mosquitto.h"
#include "mosquitto_broker.h"
static int create_tables(struct mosquitto_sqlite *ms)
{
int rc;
rc = sqlite3_exec(ms->db,
"CREATE TABLE IF NOT EXISTS msgs "
"("
"store_id INT64 PRIMARY KEY,"
"expiry_time INT64,"
"topic STRING NOT NULL,"
"payload BLOB,"
"source_id STRING,"
"source_username STRING,"
"payloadlen INTEGER,"
"source_mid INTEGER,"
"source_port INTEGER,"
"qos INTEGER,"
"retain INTEGER,"
"properties STRING"
");",
NULL, NULL, NULL);
if(rc) goto fail;
rc = sqlite3_exec(ms->db,
"CREATE TABLE IF NOT EXISTS retains "
"("
"topic STRING PRIMARY KEY,"
"store_id INT64"
//"FOREIGN KEY (store_id) REFERENCES msg_store(store_id) "
//"ON DELETE CASCADE"
");",
NULL, NULL, NULL);
if(rc) goto fail;
rc = sqlite3_exec(ms->db,
"CREATE TABLE IF NOT EXISTS clients "
"("
"client_id TEXT PRIMARY KEY,"
"username TEXT,"
"will_delay_time INT64,"
"session_expiry_time INT64,"
"listener_port INT,"
"max_packet_size INT,"
"max_qos INT,"
"retain_available INT,"
"session_expiry_interval INT,"
"will_delay_interval INT"
");",
NULL, NULL, NULL);
if(rc) goto fail;
rc = sqlite3_exec(ms->db,
"CREATE TABLE IF NOT EXISTS subscriptions "
"("
"client_id TEXT NOT NULL,"
"topic TEXT NOT NULL,"
"subscription_options INTEGER,"
"subscription_identifier INTEGER,"
"PRIMARY KEY (client_id, topic) "
");",
NULL, NULL, NULL);
if(rc) goto fail;
rc = sqlite3_exec(ms->db,
"CREATE TABLE IF NOT EXISTS client_msgs "
"("
"client_id TEXT NOT NULL,"
"store_id INT64,"
"dup INTEGER,"
"direction INTEGER,"
"mid INTEGER,"
"qos INTEGER,"
"retain INTEGER,"
"state INTEGER"
//"state INTEGER,"
//"FOREIGN KEY (client_id) REFERENCES clients(client_id) "
//"ON DELETE CASCADE,"
//"FOREIGN KEY (store_id) REFERENCES msg_store(store_id) "
//"ON DELETE CASCADE"
");",
NULL, NULL, NULL);
if(rc) goto fail;
rc = sqlite3_exec(ms->db,
"CREATE INDEX IF NOT EXISTS client_msgs_client_id ON client_msgs(client_id);",
NULL, NULL, NULL);
if(rc) goto fail;
rc = sqlite3_exec(ms->db,
"CREATE INDEX IF NOT EXISTS client_msgs_store_id ON client_msgs(store_id);",
NULL, NULL, NULL);
if(rc) goto fail;
return 0;
fail:
mosquitto_log_printf(MOSQ_LOG_ERR, "Sqlite persistence: Error creating tables: %s", sqlite3_errstr(rc));
sqlite3_close(ms->db);
ms->db = NULL;
return 1;
}
static int prepare_statements(struct mosquitto_sqlite *ms)
{
int rc;
/* Subscriptions */
rc = sqlite3_prepare_v3(ms->db,
"INSERT OR REPLACE INTO subscriptions "
"(client_id, topic, subscription_options, subscription_identifier) "
"VALUES (?,?,?,?)",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->subscription_add_stmt, NULL);
if(rc) goto fail;
rc = sqlite3_prepare_v3(ms->db,
"DELETE FROM subscriptions WHERE client_id=? and topic=?",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->subscription_remove_stmt, NULL);
if(rc) goto fail;
rc = sqlite3_prepare_v3(ms->db,
"DELETE FROM subscriptions WHERE client_id=?",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->subscription_clear_stmt, NULL);
if(rc) goto fail;
/* Clients */
rc = sqlite3_prepare_v3(ms->db,
"INSERT OR REPLACE INTO clients "
"(client_id, username, will_delay_time, session_expiry_time, "
"listener_port, max_packet_size, max_qos, retain_available, "
"session_expiry_interval, will_delay_interval) "
"VALUES(?,?,?,?,?,?,?,?,?,?)",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->client_add_stmt, NULL);
if(rc) goto fail;
rc = sqlite3_prepare_v3(ms->db,
"DELETE FROM clients WHERE client_id=?",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->client_remove_stmt, NULL);
if(rc) goto fail;
rc = sqlite3_prepare_v3(ms->db,
"UPDATE clients SET session_expiry_time=?, will_delay_time=? "
"WHERE client_id=?",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->client_update_stmt, NULL);
if(rc) goto fail;
/* Client messages */
rc = sqlite3_prepare_v3(ms->db,
"INSERT INTO client_msgs "
"(client_id,store_id,dup,direction,mid,qos,retain,state) "
"VALUES(?,?,?,?,?,?,?,?)",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->client_msg_add_stmt, NULL);
if(rc) goto fail;
rc = sqlite3_prepare_v3(ms->db,
"DELETE FROM client_msgs WHERE client_id=? AND store_id=? AND direction=?",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->client_msg_remove_stmt, NULL);
if(rc) goto fail;
rc = sqlite3_prepare_v3(ms->db,
"UPDATE client_msgs SET state=?,dup=? WHERE client_id=? AND store_id=?",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->client_msg_update_stmt, NULL);
if(rc) goto fail;
rc = sqlite3_prepare_v3(ms->db,
"DELETE FROM client_msgs WHERE client_id=? AND direction=?",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->client_msg_clear_stmt, NULL);
if(rc) goto fail;
/* Message store */
rc = sqlite3_prepare_v3(ms->db,
"INSERT INTO msgs "
"(store_id, expiry_time, topic, payload, source_id, source_username, "
"payloadlen, source_mid, source_port, qos, retain, properties) "
"VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->msg_add_stmt, NULL);
if(rc) goto fail;
rc = sqlite3_prepare_v3(ms->db,
"DELETE FROM msgs WHERE store_id=?",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->msg_remove_stmt, NULL);
if(rc) goto fail;
rc = sqlite3_prepare_v3(ms->db,
"SELECT store_id, expiry_time, topic, payload, source_id, source_username, "
"payloadlen, source_mid, source_port, qos, retain, properties "
"FROM msgs WHERE store_id=?",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->msg_load_stmt, NULL);
if(rc) goto fail;
/* Retains */
rc = sqlite3_prepare_v3(ms->db,
"INSERT OR REPLACE INTO retains "
"(topic, store_id)"
"VALUES(?,?)",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->retain_add_stmt, NULL);
if(rc) goto fail;
rc = sqlite3_prepare_v3(ms->db,
"DELETE FROM retains WHERE topic=?",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->retain_remove_stmt, NULL);
if(rc) goto fail;
return 0;
fail:
mosquitto_log_printf(MOSQ_LOG_ERR, "Sqlite persistence: Error preparing statements: %s", sqlite3_errstr(rc));
sqlite3_close(ms->db);
ms->db = NULL;
return 1;
}
int persist_sqlite__init(struct mosquitto_sqlite *ms)
{
int rc;
char buf[50];
rc = sqlite3_open_v2(ms->db_file, &ms->db, SQLITE_OPEN_READWRITE|SQLITE_OPEN_CREATE, NULL);
if(rc != SQLITE_OK){
mosquitto_log_printf(MOSQ_LOG_ERR, "Sqlite persistence: Error opening %s: %s",
ms->db_file, sqlite3_errstr(rc));
return MOSQ_ERR_UNKNOWN;
}
rc = sqlite3_exec(ms->db, "PRAGMA journal_mode=WAL;", NULL, NULL, NULL);
if(rc) goto fail;
rc = sqlite3_exec(ms->db, "PRAGMA foreign_keys = ON;", NULL, NULL, NULL);
if(rc) goto fail;
rc = sqlite3_exec(ms->db, "PRAGMA page_size=32768;", NULL, NULL, NULL);
if(rc) goto fail;
snprintf(buf, sizeof(buf), "PRAGMA synchronous=%d;", ms->synchronous);
rc = sqlite3_exec(ms->db, buf, NULL, NULL, NULL);
if(rc) goto fail;
rc = create_tables(ms);
if(rc) return rc;
rc = prepare_statements(ms);
if(rc) return rc;
return MOSQ_ERR_SUCCESS;
fail:
mosquitto_log_printf(MOSQ_LOG_ERR, "Sqlite persistence: Error opening database: %s", sqlite3_errstr(rc));
return MOSQ_ERR_UNKNOWN;
}
void persist_sqlite__cleanup(struct mosquitto_sqlite *ms)
{
sqlite3_finalize(ms->client_add_stmt);
sqlite3_finalize(ms->client_remove_stmt);
sqlite3_finalize(ms->client_update_stmt);
sqlite3_finalize(ms->subscription_add_stmt);
sqlite3_finalize(ms->subscription_remove_stmt);
sqlite3_finalize(ms->subscription_clear_stmt);
sqlite3_finalize(ms->client_msg_add_stmt);
sqlite3_finalize(ms->client_msg_remove_stmt);
sqlite3_finalize(ms->client_msg_update_stmt);
sqlite3_finalize(ms->client_msg_clear_stmt);
sqlite3_finalize(ms->msg_add_stmt);
sqlite3_finalize(ms->msg_remove_stmt);
sqlite3_finalize(ms->msg_load_stmt);
sqlite3_finalize(ms->retain_add_stmt);
sqlite3_finalize(ms->retain_remove_stmt);
if(ms->db){
sqlite3_close(ms->db);
ms->db = NULL;
}
}
+262
View File
@@ -0,0 +1,262 @@
/*
Copyright (c) 2021 Roger Light <roger@atchoo.org>
All rights reserved. This program and the accompanying materials
are made available under the terms of the Eclipse Public License 2.0
and Eclipse Distribution License v1.0 which accompany this distribution.
The Eclipse Public License is available at
https://www.eclipse.org/legal/epl-2.0/
and the Eclipse Distribution License is available at
http://www.eclipse.org/org/documents/edl-v10.php.
SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
Contributors:
Roger Light - initial implementation and documentation.
*/
#include <string.h>
#include <sqlite3.h>
#include <stdlib.h>
#include <cjson/cJSON.h>
#include "mqtt_protocol.h"
#include "mosquitto.h"
#include "mosquitto_broker.h"
#include "persist_sqlite.h"
static char *properties_to_json(const mosquitto_property *properties)
{
cJSON *array, *obj;
char *json_str, *name, *value;
uint8_t i8;
uint16_t i16;
uint32_t i32;
int propid;
if(!properties) return NULL;
array = cJSON_CreateArray();
if(!array) return NULL;
do{
propid = mosquitto_property_identifier(properties);
obj = cJSON_CreateObject();
if(!obj){
cJSON_Delete(array);
return NULL;
}
cJSON_AddItemToArray(array, obj);
/* identifier, (key), value */
if(cJSON_AddStringToObject(obj,
"identifier",
mosquitto_property_identifier_to_string(propid)) == NULL
){
cJSON_Delete(array);
return NULL;
}
switch(propid){
case MQTT_PROP_PAYLOAD_FORMAT_INDICATOR:
case MQTT_PROP_REQUEST_PROBLEM_INFORMATION:
case MQTT_PROP_REQUEST_RESPONSE_INFORMATION:
case MQTT_PROP_MAXIMUM_QOS:
case MQTT_PROP_RETAIN_AVAILABLE:
case MQTT_PROP_WILDCARD_SUB_AVAILABLE:
case MQTT_PROP_SUBSCRIPTION_ID_AVAILABLE:
case MQTT_PROP_SHARED_SUB_AVAILABLE:
/* byte */
mosquitto_property_read_byte(properties, propid, &i8, false);
if(cJSON_AddNumberToObject(obj, "value", i8) == NULL){
cJSON_Delete(array);
return NULL;
}
break;
case MQTT_PROP_SERVER_KEEP_ALIVE:
case MQTT_PROP_RECEIVE_MAXIMUM:
case MQTT_PROP_TOPIC_ALIAS_MAXIMUM:
case MQTT_PROP_TOPIC_ALIAS:
/* 2 byte */
mosquitto_property_read_int16(properties, propid, &i16, false);
if(cJSON_AddNumberToObject(obj, "value", i16) == NULL){
cJSON_Delete(array);
return NULL;
}
break;
case MQTT_PROP_MESSAGE_EXPIRY_INTERVAL:
case MQTT_PROP_SESSION_EXPIRY_INTERVAL:
case MQTT_PROP_WILL_DELAY_INTERVAL:
case MQTT_PROP_MAXIMUM_PACKET_SIZE:
/* 4 byte */
mosquitto_property_read_int32(properties, propid, &i32, false);
if(cJSON_AddNumberToObject(obj, "value", i32) == NULL){
cJSON_Delete(array);
return NULL;
}
break;
case MQTT_PROP_SUBSCRIPTION_IDENTIFIER:
/* var byte */
mosquitto_property_read_varint(properties, propid, &i32, false);
if(cJSON_AddNumberToObject(obj, "value", i32) == NULL){
cJSON_Delete(array);
return NULL;
}
break;
case MQTT_PROP_CONTENT_TYPE:
case MQTT_PROP_RESPONSE_TOPIC:
case MQTT_PROP_ASSIGNED_CLIENT_IDENTIFIER:
case MQTT_PROP_AUTHENTICATION_METHOD:
case MQTT_PROP_RESPONSE_INFORMATION:
case MQTT_PROP_SERVER_REFERENCE:
case MQTT_PROP_REASON_STRING:
/* str */
mosquitto_property_read_string(properties, propid, &value, false);
if(cJSON_AddStringToObject(obj, "value", value) == NULL){
free(value);
cJSON_Delete(array);
return NULL;
}
free(value);
break;
case MQTT_PROP_CORRELATION_DATA:
case MQTT_PROP_AUTHENTICATION_DATA:
/* bin */
break;
case MQTT_PROP_USER_PROPERTY:
/* pair */
mosquitto_property_read_string_pair(properties, propid, &name, &value, false);
if(cJSON_AddStringToObject(obj, "name", name) == NULL
|| cJSON_AddStringToObject(obj, "value", value) == NULL){
free(name);
free(value);
cJSON_Delete(array);
return NULL;
}
free(name);
free(value);
break;
default:
break;
}
properties = mosquitto_property_next(properties);
}while(properties);
json_str = cJSON_PrintUnformatted(array);
cJSON_Delete(array);
return json_str;
}
int persist_sqlite__msg_add_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_evt_persist_msg *ed = event_data;
struct mosquitto_sqlite *ms = userdata;
int rc = MOSQ_ERR_UNKNOWN;
char *str = NULL;
UNUSED(event);
rc = 0;
rc += sqlite3_bind_int64(ms->msg_add_stmt, 1, (int64_t)ed->store_id);
rc += sqlite3_bind_int64(ms->msg_add_stmt, 2, ed->expiry_time);
rc += sqlite3_bind_text(ms->msg_add_stmt, 3, ed->topic, (int)strlen(ed->topic), SQLITE_STATIC);
if(ed->payload){
rc += sqlite3_bind_blob(ms->msg_add_stmt, 4, ed->payload, (int)ed->payloadlen, SQLITE_STATIC);
}else{
rc += sqlite3_bind_null(ms->msg_add_stmt, 4);
}
if(ed->source_id){
rc += sqlite3_bind_text(ms->msg_add_stmt, 5, ed->source_id, (int)strlen(ed->source_id), SQLITE_STATIC);
}else{
rc += sqlite3_bind_null(ms->msg_add_stmt, 5);
}
if(ed->source_username){
rc += sqlite3_bind_text(ms->msg_add_stmt, 6, ed->source_username, (int)strlen(ed->source_username), SQLITE_STATIC);
}else{
rc += sqlite3_bind_null(ms->msg_add_stmt, 6);
}
rc += sqlite3_bind_int(ms->msg_add_stmt, 7, (int)ed->payloadlen);
rc += sqlite3_bind_int(ms->msg_add_stmt, 8, ed->source_mid);
rc += sqlite3_bind_int(ms->msg_add_stmt, 9, ed->source_port);
rc += sqlite3_bind_int(ms->msg_add_stmt, 10, ed->qos);
rc += sqlite3_bind_int(ms->msg_add_stmt, 11, ed->retain);
if(ed->properties){
str = properties_to_json(ed->properties);
}
if(str){
rc += sqlite3_bind_text(ms->msg_add_stmt, 12, str, (int)strlen(str), SQLITE_STATIC);
}else{
rc += sqlite3_bind_null(ms->msg_add_stmt, 12);
}
if(rc == 0){
rc = sqlite3_step(ms->msg_add_stmt);
if(rc == SQLITE_DONE){
rc = MOSQ_ERR_SUCCESS;
}else{
rc = MOSQ_ERR_UNKNOWN;
}
}
sqlite3_reset(ms->msg_add_stmt);
free(str);
return rc;
}
int persist_sqlite__msg_remove_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_evt_persist_msg *ed = event_data;
struct mosquitto_sqlite *ms = userdata;
int rc = 1;
UNUSED(event);
if(sqlite3_bind_int64(ms->msg_remove_stmt, 1, (int64_t)ed->store_id) == SQLITE_OK){
rc = sqlite3_step(ms->msg_remove_stmt);
if(rc == SQLITE_DONE){
rc = MOSQ_ERR_SUCCESS;
}else{
rc = MOSQ_ERR_UNKNOWN;
}
}
sqlite3_reset(ms->msg_remove_stmt);
return rc;
}
int persist_sqlite__msg_load_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_evt_persist_msg *msg = event_data;
struct mosquitto_sqlite *ms = userdata;
UNUSED(event);
if(sqlite3_bind_int64(ms->msg_load_stmt, 1, (int64_t)msg->store_id) == SQLITE_OK){
if(sqlite3_step(ms->msg_load_stmt) == SQLITE_ROW){
msg->expiry_time = (time_t)sqlite3_column_int64(ms->msg_load_stmt, 1);
msg->topic = (char *)sqlite3_column_text(ms->msg_load_stmt, 2);
msg->payload = (void *)sqlite3_column_blob(ms->msg_load_stmt, 3);
msg->source_id = (char *)sqlite3_column_text(ms->msg_load_stmt, 4);
msg->source_username = (char *)sqlite3_column_text(ms->msg_load_stmt, 5);
msg->payloadlen = (uint32_t)sqlite3_column_int(ms->msg_load_stmt, 6);
msg->source_mid = (uint16_t)sqlite3_column_int(ms->msg_load_stmt, 7);
msg->source_port = (uint16_t)sqlite3_column_int(ms->msg_load_stmt, 8);
msg->qos = (uint8_t)sqlite3_column_int(ms->msg_load_stmt, 9);
msg->retain = sqlite3_column_int(ms->msg_load_stmt, 10);
mosquitto_persist_msg_add(msg);
}
}
sqlite3_finalize(ms->msg_load_stmt);
return MOSQ_ERR_SUCCESS;
}
+68
View File
@@ -0,0 +1,68 @@
/*
Copyright (c) 2021 Roger Light <roger@atchoo.org>
All rights reserved. This program and the accompanying materials
are made available under the terms of the Eclipse Public License 2.0
and Eclipse Distribution License v1.0 which accompany this distribution.
The Eclipse Public License is available at
https://www.eclipse.org/legal/epl-2.0/
and the Eclipse Distribution License is available at
http://www.eclipse.org/org/documents/edl-v10.php.
SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
Contributors:
Roger Light - initial implementation and documentation.
*/
#ifndef PERSIST_SQLITE_H
#define PERSIST_SQLITE_H
#include <sqlite3.h>
#ifndef UNUSED
# define UNUSED(A) (void)(A)
#endif
struct mosquitto_sqlite {
char *db_file;
sqlite3 *db;
sqlite3_stmt *client_add_stmt;
sqlite3_stmt *client_remove_stmt;
sqlite3_stmt *client_update_stmt;
sqlite3_stmt *subscription_add_stmt;
sqlite3_stmt *subscription_remove_stmt;
sqlite3_stmt *subscription_clear_stmt;
sqlite3_stmt *client_msg_add_stmt;
sqlite3_stmt *client_msg_remove_stmt;
sqlite3_stmt *client_msg_update_stmt;
sqlite3_stmt *client_msg_clear_stmt;
sqlite3_stmt *msg_add_stmt;
sqlite3_stmt *msg_remove_stmt;
sqlite3_stmt *msg_load_stmt;
sqlite3_stmt *retain_add_stmt;
sqlite3_stmt *retain_remove_stmt;
int synchronous;
};
int persist_sqlite__init(struct mosquitto_sqlite *ms);
void persist_sqlite__cleanup(struct mosquitto_sqlite *ms);
int persist_sqlite__restore_cb(int event, void *event_data, void *userdata);
int persist_sqlite__client_add_cb(int event, void *event_data, void *userdata);
int persist_sqlite__client_update_cb(int event, void *event_data, void *userdata);
int persist_sqlite__client_remove_cb(int event, void *event_data, void *userdata);
int persist_sqlite__client_msg_add_cb(int event, void *event_data, void *userdata);
int persist_sqlite__client_msg_clear_cb(int event, void *event_data, void *userdata);
int persist_sqlite__client_msg_remove_cb(int event, void *event_data, void *userdata);
int persist_sqlite__client_msg_update_cb(int event, void *event_data, void *userdata);
int persist_sqlite__msg_add_cb(int event, void *event_data, void *userdata);
int persist_sqlite__msg_load_cb(int event, void *event_data, void *userdata);
int persist_sqlite__msg_remove_cb(int event, void *event_data, void *userdata);
int persist_sqlite__retain_add_cb(int event, void *event_data, void *userdata);
int persist_sqlite__retain_remove_cb(int event, void *event_data, void *userdata);
int persist_sqlite__subscription_add_cb(int event, void *event_data, void *userdata);
int persist_sqlite__subscription_remove_cb(int event, void *event_data, void *userdata);
#endif
+161
View File
@@ -0,0 +1,161 @@
/*
Copyright (c) 2021 Roger Light <roger@atchoo.org>
All rights reserved. This program and the accompanying materials
are made available under the terms of the Eclipse Public License 2.0
and Eclipse Distribution License v1.0 which accompany this distribution.
The Eclipse Public License is available at
https://www.eclipse.org/legal/epl-2.0/
and the Eclipse Distribution License is available at
http://www.eclipse.org/org/documents/edl-v10.php.
SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
Contributors:
Roger Light - initial implementation and documentation.
*/
#include "config.h"
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include "mosquitto.h"
#include "mosquitto_broker.h"
#include "mosquitto_plugin.h"
#include "mqtt_protocol.h"
#include "persist_sqlite.h"
static mosquitto_plugin_id_t *plg_id = NULL;
static struct mosquitto_sqlite plg_data;
int mosquitto_plugin_version(int supported_version_count, const int *supported_versions)
{
int i;
for(i=0; i<supported_version_count; i++){
if(supported_versions[i] == 5){
return 5;
}
}
return -1;
}
int mosquitto_plugin_init(mosquitto_plugin_id_t *identifier, void **user_data, struct mosquitto_opt *options, int option_count)
{
int i;
int rc;
UNUSED(user_data);
memset(&plg_data, 0,sizeof(struct mosquitto_sqlite));
for(i=0; i<option_count; i++){
if(!strcasecmp(options[i].key, "db_file")){
plg_data.db_file = mosquitto_strdup(options[i].value);
if(plg_data.db_file == NULL){
return MOSQ_ERR_NOMEM;
}
break;
}else if(!strcasecmp(options[i].key, "sync")){
if(!strcasecmp(options[i].value, "extra")){
plg_data.synchronous = 3;
}else if(!strcasecmp(options[i].value, "full")){
plg_data.synchronous = 2;
}else if(!strcasecmp(options[i].value, "normal")){
plg_data.synchronous = 1;
}else if(!strcasecmp(options[i].value, "off")){
plg_data.synchronous = 0;
}else{
mosquitto_log_printf(MOSQ_LOG_ERR, "Sqlite persistence: Invalid plugin_opt_sync value '%s'.", options[i].value);
return MOSQ_ERR_INVAL;
}
}
}
if(plg_data.db_file == NULL){
mosquitto_log_printf(MOSQ_LOG_WARNING, "Warning: Sqlite persistence plugin has no plugin_opt_db_file defined. The plugin will not be activated.");
return MOSQ_ERR_SUCCESS;
}
rc = persist_sqlite__init(&plg_data);
if(rc) return rc;
plg_id = identifier;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_RESTORE, persist_sqlite__restore_cb, NULL, &plg_data);
if(rc) goto fail;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_MSG_ADD, persist_sqlite__msg_add_cb, NULL, &plg_data);
if(rc) goto fail;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_MSG_DELETE, persist_sqlite__msg_remove_cb, NULL, &plg_data);
if(rc) goto fail;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_MSG_LOAD, persist_sqlite__msg_load_cb, NULL, &plg_data);
if(rc) goto fail;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_RETAIN_ADD, persist_sqlite__retain_add_cb, NULL, &plg_data);
if(rc) goto fail;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_RETAIN_DELETE, persist_sqlite__retain_remove_cb, NULL, &plg_data);
if(rc) goto fail;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_CLIENT_ADD, persist_sqlite__client_add_cb, NULL, &plg_data);
if(rc) goto fail;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_CLIENT_DELETE, persist_sqlite__client_remove_cb, NULL, &plg_data);
if(rc) goto fail;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_CLIENT_UPDATE, persist_sqlite__client_update_cb, NULL, &plg_data);
if(rc) goto fail;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_SUBSCRIPTION_ADD, persist_sqlite__subscription_add_cb, NULL, &plg_data);
if(rc) goto fail;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_SUBSCRIPTION_DELETE, persist_sqlite__subscription_remove_cb, NULL, &plg_data);
if(rc) goto fail;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_CLIENT_MSG_ADD, persist_sqlite__client_msg_add_cb, NULL, &plg_data);
if(rc) goto fail;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_CLIENT_MSG_DELETE, persist_sqlite__client_msg_remove_cb, NULL, &plg_data);
if(rc) goto fail;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_CLIENT_MSG_UPDATE, persist_sqlite__client_msg_update_cb, NULL, &plg_data);
if(rc) goto fail;
rc = mosquitto_callback_register(plg_id, MOSQ_EVT_PERSIST_CLIENT_MSG_CLEAR, persist_sqlite__client_msg_clear_cb, NULL, &plg_data);
if(rc) goto fail;
return MOSQ_ERR_SUCCESS;
fail:
if(rc == MOSQ_ERR_NOT_SUPPORTED){
mosquitto_log_printf(MOSQ_LOG_ERR, "Sqlite persistence: Unable to register plugin: broker doesn't support persistence plugins, please upgrade to 2.1 or higher");
}else if(rc == MOSQ_ERR_NOMEM){
mosquitto_log_printf(MOSQ_LOG_ERR, "Sqlite persistence: Unable to register plugin: out of memory");
}else{
mosquitto_log_printf(MOSQ_LOG_ERR, "Sqlite persistence: Unable to register plugin (%d)", rc);
}
mosquitto_plugin_cleanup(NULL, NULL, 0);
return rc;
}
int mosquitto_plugin_cleanup(void *user_data, struct mosquitto_opt *options, int option_count)
{
UNUSED(user_data);
UNUSED(options);
UNUSED(option_count);
if(plg_id){
mosquitto_callback_unregister(plg_id, MOSQ_EVT_PERSIST_RESTORE, persist_sqlite__restore_cb, NULL);
mosquitto_callback_unregister(plg_id, MOSQ_EVT_PERSIST_MSG_ADD, persist_sqlite__msg_add_cb, NULL);
mosquitto_callback_unregister(plg_id, MOSQ_EVT_PERSIST_MSG_DELETE, persist_sqlite__msg_remove_cb, NULL);
mosquitto_callback_unregister(plg_id, MOSQ_EVT_PERSIST_MSG_LOAD, persist_sqlite__msg_load_cb, NULL);
mosquitto_callback_unregister(plg_id, MOSQ_EVT_PERSIST_RETAIN_ADD, persist_sqlite__retain_add_cb, NULL);
mosquitto_callback_unregister(plg_id, MOSQ_EVT_PERSIST_RETAIN_DELETE, persist_sqlite__retain_remove_cb, NULL);
mosquitto_callback_unregister(plg_id, MOSQ_EVT_PERSIST_CLIENT_ADD, persist_sqlite__client_add_cb, NULL);
mosquitto_callback_unregister(plg_id, MOSQ_EVT_PERSIST_CLIENT_DELETE, persist_sqlite__client_remove_cb, NULL);
mosquitto_callback_unregister(plg_id, MOSQ_EVT_PERSIST_SUBSCRIPTION_ADD, persist_sqlite__subscription_add_cb, NULL);
mosquitto_callback_unregister(plg_id, MOSQ_EVT_PERSIST_SUBSCRIPTION_DELETE, persist_sqlite__subscription_remove_cb, NULL);
mosquitto_callback_unregister(plg_id, MOSQ_EVT_PERSIST_CLIENT_MSG_ADD, persist_sqlite__client_msg_add_cb, NULL);
mosquitto_callback_unregister(plg_id, MOSQ_EVT_PERSIST_CLIENT_MSG_DELETE, persist_sqlite__client_msg_remove_cb, NULL);
mosquitto_callback_unregister(plg_id, MOSQ_EVT_PERSIST_CLIENT_MSG_UPDATE, persist_sqlite__client_msg_update_cb, NULL);
mosquitto_callback_unregister(plg_id, MOSQ_EVT_PERSIST_CLIENT_MSG_CLEAR, persist_sqlite__client_msg_clear_cb, NULL);
}
mosquitto_free(plg_data.db_file);
persist_sqlite__cleanup(&plg_data);
memset(&plg_data, 0, sizeof(struct mosquitto_sqlite));
return MOSQ_ERR_SUCCESS;
}
+373
View File
@@ -0,0 +1,373 @@
/*
Copyright (c) 2021 Roger Light <roger@atchoo.org>
All rights reserved. This program and the accompanying materials
are made available under the terms of the Eclipse Public License 2.0
and Eclipse Distribution License v1.0 which accompany this distribution.
The Eclipse Public License is available at
https://www.eclipse.org/legal/epl-2.0/
and the Eclipse Distribution License is available at
http://www.eclipse.org/org/documents/edl-v10.php.
SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
Contributors:
Roger Light - initial implementation and documentation.
*/
#include <stdlib.h>
#include <string.h>
#include <sqlite3.h>
#include <cjson/cJSON.h>
#include "mosquitto.h"
#include "mosquitto_broker.h"
#include "mqtt_protocol.h"
#include "persist_sqlite.h"
static mosquitto_property *json_to_properties(const char *json)
{
mosquitto_property *properties = NULL;
cJSON *array, *obj, *j_id, *j_value, *j_name;
int propid, proptype;
if(!json) return NULL;
array = cJSON_Parse(json);
if(!array) return NULL;
if(!cJSON_IsArray(array)){
cJSON_Delete(array);
return NULL;
}
cJSON_ArrayForEach(obj, array){
j_id = cJSON_GetObjectItem(obj, "identifier");
j_name = cJSON_GetObjectItem(obj, "name");
j_value = cJSON_GetObjectItem(obj, "value");
if(!j_id || !cJSON_IsString(j_id) || !j_value){
continue;
}
if(mosquitto_string_to_property_info(j_id->valuestring, &propid, &proptype)){
continue;
}
switch(proptype){
case MQTT_PROP_TYPE_BYTE:
if(!cJSON_IsNumber(j_value)) continue;
mosquitto_property_add_byte(&properties, propid, (uint8_t)j_value->valueint);
break;
case MQTT_PROP_TYPE_INT16:
if(!cJSON_IsNumber(j_value)) continue;
mosquitto_property_add_int16(&properties, propid, (uint16_t)j_value->valueint);
break;
case MQTT_PROP_TYPE_INT32:
if(!cJSON_IsNumber(j_value)) continue;
mosquitto_property_add_int32(&properties, propid, (uint32_t)j_value->valueint);
break;
case MQTT_PROP_TYPE_VARINT:
if(!cJSON_IsNumber(j_value)) continue;
mosquitto_property_add_varint(&properties, propid, (uint32_t)j_value->valueint);
break;
case MQTT_PROP_TYPE_BINARY:
break;
case MQTT_PROP_TYPE_STRING:
if(!cJSON_IsString(j_value)) continue;
mosquitto_property_add_string(&properties, propid, j_value->valuestring);
break;
case MQTT_PROP_TYPE_STRING_PAIR:
if(!cJSON_IsString(j_value)) continue;
if(!j_name || !cJSON_IsString(j_value)) continue;
mosquitto_property_add_string_pair(&properties, propid, j_name->valuestring, j_value->valuestring);
break;
}
}
cJSON_Delete(array);
return properties;
}
static int client_restore(struct mosquitto_sqlite *ms)
{
sqlite3_stmt *stmt;
int rc;
struct mosquitto_evt_persist_client client;
long count = 0, failed = 0;
const char *str;
memset(&client, 0, sizeof(client));
rc = sqlite3_prepare_v2(ms->db,
"SELECT client_id,username,will_delay_time,session_expiry_time,"
"listener_port,max_packet_size,max_qos,"
"retain_available,session_expiry_interval,will_delay_interval "
"FROM clients",
-1, &stmt, NULL);
if(rc != SQLITE_OK){
mosquitto_log_printf(MOSQ_LOG_ERR, "sqlite: Error restoring clients: %s", sqlite3_errstr(rc));
return MOSQ_ERR_UNKNOWN;
}
while(sqlite3_step(stmt) == SQLITE_ROW){
str = (const char *)sqlite3_column_text(stmt, 0);
if(str){
client.plugin_client_id = strdup(str);
}
str = (const char *)sqlite3_column_text(stmt, 1);
if(str){
client.plugin_username = strdup(str);
}
client.will_delay_time = (time_t)sqlite3_column_int64(stmt, 2);
client.session_expiry_time = (time_t)sqlite3_column_int64(stmt, 3);
client.listener_port = (uint16_t)sqlite3_column_int(stmt, 4);
client.max_packet_size = (uint32_t)sqlite3_column_int(stmt, 5);
client.max_qos = (uint8_t)sqlite3_column_int(stmt, 6);
client.retain_available = (bool)sqlite3_column_int(stmt, 7);
client.session_expiry_interval = (uint32_t)sqlite3_column_int(stmt, 8);
client.will_delay_interval = (uint32_t)sqlite3_column_int(stmt, 9);
rc = mosquitto_persist_client_add(&client);
if(rc == MOSQ_ERR_SUCCESS){
count++;
}else{
failed++;
}
}
sqlite3_finalize(stmt);
mosquitto_log_printf(MOSQ_LOG_INFO, "sqlite: Restored %d clients (%ld failed)", count, failed);
return rc;
}
static int subscription_restore(struct mosquitto_sqlite *ms)
{
sqlite3_stmt *stmt;
uint8_t subscription_options;
uint32_t subscription_identifier;
int rc;
const char *client_id;
const char *topic;
long count = 0, failed = 0;
rc = sqlite3_prepare_v2(ms->db,
"SELECT client_id,topic,subscription_options,subscription_identifier "
"FROM subscriptions",
-1, &stmt, NULL);
if(rc != SQLITE_OK){
mosquitto_log_printf(MOSQ_LOG_ERR, "sqlite: Error restoring subscriptions: %s", sqlite3_errstr(rc));
return MOSQ_ERR_UNKNOWN;
}
while(sqlite3_step(stmt) == SQLITE_ROW){
client_id = (const char *)sqlite3_column_text(stmt, 0);
topic = (const char *)sqlite3_column_text(stmt, 1);
subscription_options = (uint8_t)sqlite3_column_int(stmt, 2);
subscription_identifier = (uint32_t)sqlite3_column_int(stmt, 3);
rc = mosquitto_subscription_add(client_id, topic, subscription_options, subscription_identifier);
if(rc == MOSQ_ERR_SUCCESS){
count++;
}else{
failed++;
}
}
sqlite3_finalize(stmt);
mosquitto_log_printf(MOSQ_LOG_INFO, "sqlite: Restored %d subscriptions (%ld failed)", count, failed);
return MOSQ_ERR_SUCCESS;
}
static int msg_restore(struct mosquitto_sqlite *ms)
{
sqlite3_stmt *stmt;
struct mosquitto_evt_persist_msg msg;
int rc;
long count = 0, failed = 0;
const char *str;
const void *payload;
rc = sqlite3_prepare_v2(ms->db,
"SELECT store_id, expiry_time, topic, payload, source_id, source_username, payloadlen, source_mid, source_port, qos, retain, properties "
"FROM msgs",
-1, &stmt, NULL);
if(rc != SQLITE_OK){
mosquitto_log_printf(MOSQ_LOG_ERR, "sqlite: Error restoring messages: %s", sqlite3_errstr(rc));
return MOSQ_ERR_UNKNOWN;
}
while(sqlite3_step(stmt) == SQLITE_ROW){
memset(&msg, 0, sizeof(msg));
msg.store_id = (uint64_t)sqlite3_column_int64(stmt, 0);
msg.expiry_time = (time_t)sqlite3_column_int64(stmt, 1);
str = (const char *)sqlite3_column_text(stmt, 2);
if(str){
msg.plugin_topic = strdup(str);
if(!msg.plugin_topic){
failed++;
continue;
}
}
str = (const char *)sqlite3_column_text(stmt, 4);
if(str){
msg.plugin_source_id = strdup(str);
if(!msg.plugin_source_id){
free(msg.plugin_topic);
failed++;
continue;
}
}
str = (const char *)sqlite3_column_text(stmt, 5);
if(str){
msg.plugin_source_username = strdup(str);
if(!msg.plugin_source_username){
free(msg.plugin_topic);
free(msg.plugin_source_id);
failed++;
continue;
}
}
payload = (const void *)sqlite3_column_blob(stmt, 3);
msg.payloadlen = (uint32_t)sqlite3_column_int(stmt, 6);
if(payload && msg.payloadlen){
msg.plugin_payload = malloc(msg.payloadlen+1);
if(!msg.plugin_payload){
free(msg.plugin_topic);
free(msg.plugin_topic);
free(msg.plugin_source_id);
free(msg.plugin_source_username);
failed++;
continue;
}
memcpy(msg.plugin_payload, payload, msg.payloadlen);
((uint8_t *)msg.plugin_payload)[msg.payloadlen] = 0;
}
msg.source_mid = (uint16_t)sqlite3_column_int(stmt, 7);
msg.source_port = (uint16_t)sqlite3_column_int(stmt, 8);
msg.qos = (uint8_t)sqlite3_column_int(stmt, 9);
msg.retain = sqlite3_column_int(stmt, 10);
msg.plugin_properties = json_to_properties((const char *)sqlite3_column_text(stmt, 11));
rc = mosquitto_persist_msg_add(&msg);
if(rc == MOSQ_ERR_SUCCESS){
count++;
}else{
failed++;
}
}
sqlite3_finalize(stmt);
mosquitto_log_printf(MOSQ_LOG_INFO, "sqlite: Restored %d messages (%ld failed)", count, failed);
return MOSQ_ERR_SUCCESS;
}
static int client_msg_restore(struct mosquitto_sqlite *ms)
{
sqlite3_stmt *stmt;
struct mosquitto_evt_persist_client_msg msg;
int rc;
long count = 0, failed = 0;
const char *str;
rc = sqlite3_prepare_v2(ms->db,
"SELECT client_id, store_id, dup, direction, mid, qos, retain, state "
"FROM client_msgs ORDER BY rowid",
-1, &stmt, NULL);
if(rc != SQLITE_OK){
mosquitto_log_printf(MOSQ_LOG_ERR, "sqlite: Error restoring client messages: %s", sqlite3_errstr(rc));
return MOSQ_ERR_UNKNOWN;
}
memset(&msg, 0, sizeof(msg));
while(sqlite3_step(stmt) == SQLITE_ROW){
str = (const char *)sqlite3_column_text(stmt, 0);
if(str){
msg.plugin_client_id = strdup(str);
}
msg.store_id = (uint64_t)sqlite3_column_int64(stmt, 1);
msg.dup = sqlite3_column_int(stmt, 2);
msg.direction = (uint8_t)sqlite3_column_int(stmt, 3);
msg.mid = (uint16_t)sqlite3_column_int(stmt, 4);
msg.qos = (uint8_t)sqlite3_column_int(stmt, 5);
msg.retain = sqlite3_column_int(stmt, 6);
msg.state = (uint8_t)sqlite3_column_int(stmt, 7);
rc = mosquitto_persist_client_msg_add(&msg);
if(rc == MOSQ_ERR_SUCCESS){
count++;
}else{
failed++;
}
}
sqlite3_finalize(stmt);
mosquitto_log_printf(MOSQ_LOG_INFO, "sqlite: Restored %d client messages (%ld failed)", count, failed);
return MOSQ_ERR_SUCCESS;
}
static int retain_restore(struct mosquitto_sqlite *ms)
{
sqlite3_stmt *stmt;
int rc;
long count = 0, failed = 0;
const char *topic;
uint64_t store_id;
rc = sqlite3_prepare_v2(ms->db,
"SELECT topic, store_id "
"FROM retains ORDER BY topic",
-1, &stmt, NULL);
if(rc != SQLITE_OK){
mosquitto_log_printf(MOSQ_LOG_ERR, "sqlite: Error restoring retained messages: %s", sqlite3_errstr(rc));
return MOSQ_ERR_UNKNOWN;
}
while(sqlite3_step(stmt) == SQLITE_ROW){
topic = (const char *)sqlite3_column_text(stmt, 0);
if(!topic){
failed++;
continue;
}
store_id = (uint64_t)sqlite3_column_int64(stmt, 1);
rc = mosquitto_persist_retain_add(topic, store_id);
if(rc == MOSQ_ERR_SUCCESS){
count++;
}else{
failed++;
}
}
sqlite3_finalize(stmt);
mosquitto_log_printf(MOSQ_LOG_INFO, "sqlite: Restored %d retained messages (%ld failed)", count, failed);
return MOSQ_ERR_SUCCESS;
}
int persist_sqlite__restore_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_sqlite *ms = userdata;
UNUSED(event);
UNUSED(event_data);
if(msg_restore(ms)) return MOSQ_ERR_UNKNOWN;
if(retain_restore(ms)) return MOSQ_ERR_UNKNOWN;
if(client_restore(ms)) return MOSQ_ERR_UNKNOWN;
if(subscription_restore(ms)) return MOSQ_ERR_UNKNOWN;
if(client_msg_restore(ms)) return MOSQ_ERR_UNKNOWN;
return 0;
}
+71
View File
@@ -0,0 +1,71 @@
/*
Copyright (c) 2021 Roger Light <roger@atchoo.org>
All rights reserved. This program and the accompanying materials
are made available under the terms of the Eclipse Public License 2.0
and Eclipse Distribution License v1.0 which accompany this distribution.
The Eclipse Public License is available at
https://www.eclipse.org/legal/epl-2.0/
and the Eclipse Distribution License is available at
http://www.eclipse.org/org/documents/edl-v10.php.
SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
Contributors:
Roger Light - initial implementation and documentation.
*/
#include <string.h>
#include <sqlite3.h>
#include "mosquitto.h"
#include "mosquitto_broker.h"
#include "persist_sqlite.h"
int persist_sqlite__retain_add_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_evt_persist_retain *ed = event_data;
struct mosquitto_sqlite *ms = userdata;
int rc = MOSQ_ERR_UNKNOWN;
UNUSED(event);
if(sqlite3_bind_text(ms->retain_add_stmt, 1, ed->topic, (int)strlen(ed->topic), SQLITE_STATIC) == SQLITE_OK
&& sqlite3_bind_int64(ms->retain_add_stmt, 2, (int64_t)ed->store_id) == SQLITE_OK
){
rc = sqlite3_step(ms->retain_add_stmt);
if(rc == SQLITE_DONE){
rc = MOSQ_ERR_SUCCESS;
}else{
rc = MOSQ_ERR_UNKNOWN;
}
}
sqlite3_reset(ms->retain_add_stmt);
return rc;
}
int persist_sqlite__retain_remove_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_evt_persist_retain *ed = event_data;
struct mosquitto_sqlite *ms = userdata;
int rc = 1;
UNUSED(event);
if(sqlite3_bind_text(ms->retain_remove_stmt, 1,
ed->topic, (int)strlen(ed->topic), SQLITE_STATIC) == SQLITE_OK){
rc = sqlite3_step(ms->retain_remove_stmt);
if(rc == SQLITE_DONE){
rc = MOSQ_ERR_SUCCESS;
}else{
rc = MOSQ_ERR_UNKNOWN;
}
}
sqlite3_reset(ms->retain_remove_stmt);
return rc;
}
+86
View File
@@ -0,0 +1,86 @@
/*
Copyright (c) 2021 Roger Light <roger@atchoo.org>
All rights reserved. This program and the accompanying materials
are made available under the terms of the Eclipse Public License 2.0
and Eclipse Distribution License v1.0 which accompany this distribution.
The Eclipse Public License is available at
https://www.eclipse.org/legal/epl-2.0/
and the Eclipse Distribution License is available at
http://www.eclipse.org/org/documents/edl-v10.php.
SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
Contributors:
Roger Light - initial implementation and documentation.
*/
#include <string.h>
#include <sqlite3.h>
#include "mosquitto.h"
#include "mosquitto_broker.h"
#include "persist_sqlite.h"
int persist_sqlite__subscription_add_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_evt_persist_subscription *ed = event_data;
struct mosquitto_sqlite *ms = userdata;
int rc = MOSQ_ERR_UNKNOWN;
UNUSED(event);
if(sqlite3_bind_text(ms->subscription_add_stmt, 1,
ed->client_id, (int)strlen(ed->client_id), SQLITE_STATIC) == SQLITE_OK){
if(sqlite3_bind_text(ms->subscription_add_stmt, 2,
ed->topic, (int)strlen(ed->topic), SQLITE_STATIC) == SQLITE_OK){
if(sqlite3_bind_int(ms->subscription_add_stmt, 3,
ed->subscription_options) == SQLITE_OK){
if(sqlite3_bind_int(ms->subscription_add_stmt, 4,
(int)ed->subscription_identifier) == SQLITE_OK){
rc = sqlite3_step(ms->subscription_add_stmt);
if(rc == SQLITE_DONE){
rc = MOSQ_ERR_SUCCESS;
}else{
rc = MOSQ_ERR_UNKNOWN;
}
}
}
}
}
sqlite3_reset(ms->subscription_add_stmt);
return rc;
}
int persist_sqlite__subscription_remove_cb(int event, void *event_data, void *userdata)
{
struct mosquitto_evt_persist_subscription *ed = event_data;
struct mosquitto_sqlite *ms = userdata;
int rc = 1;
UNUSED(event);
if(sqlite3_bind_text(ms->subscription_remove_stmt, 1,
ed->client_id, (int)strlen(ed->client_id), SQLITE_STATIC) == SQLITE_OK){
if(sqlite3_bind_text(ms->subscription_remove_stmt, 2,
ed->topic, (int)strlen(ed->topic), SQLITE_STATIC) == SQLITE_OK){
rc = sqlite3_step(ms->subscription_remove_stmt);
if(rc == SQLITE_DONE){
rc = MOSQ_ERR_SUCCESS;
}else{
rc = MOSQ_ERR_UNKNOWN;
}
}
}
sqlite3_reset(ms->subscription_remove_stmt);
return rc;
}
+2
View File
@@ -0,0 +1,2 @@
plugin ./mosquitto_persist_sqlite.so
plugin_opt_db_file test.sqlite3
+1
View File
@@ -0,0 +1 @@
../../src/mosquitto -c test.conf -v
+1 -1
View File
@@ -282,7 +282,7 @@ void db__msg_store_remove(struct mosquitto_msg_store *store, bool notify)
db.msg_store_count--;
db.msg_store_bytes -= store->payloadlen;
if(notify == true){
plugin_persist__handle_msg_remove(store);
plugin_persist__handle_msg_delete(store);
}
db__msg_store_free(store);
}
+42
View File
@@ -0,0 +1,42 @@
#!/usr/bin/env python3
# Check whether the sqlite plugin cleans everything up before closing - this
# means the WAL journal file will not exist when it has closed.
from mosq_test_helper import *
import json
import shutil
import sqlite_help
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
sqlite_help.init(port)
proto_ver = 4
connect_packet = mosq_test.gen_connect("sqlite-clean-shutdown", proto_ver=4)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=4)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
try:
# Check broker is running
sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=5, port=port)
sock.close()
except mosq_test.TestError:
pass
finally:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
os.remove(conf_file)
rc = sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+91
View File
@@ -0,0 +1,91 @@
#!/usr/bin/env python3
# Connect a client, start a QoS 2 flow, disconnect, restore, carry on with the
# QoS 2 flow. Is it received?
from mosq_test_helper import *
import sqlite_help
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
sqlite_help.init(port)
client_id = "persist-client-msg-in-v3-1-1"
proto_ver = 4
helper_id = "persist-client-msg-in-v3-1-1-helper"
topic = "client-msg-in/2"
qos = 2
connect_packet = mosq_test.gen_connect(client_id, proto_ver=proto_ver, clean_session=False)
connack_packet1 = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
connack_packet2 = mosq_test.gen_connack(rc=0, flags=1, proto_ver=proto_ver)
mid = 1
publish1_packet = mosq_test.gen_publish(topic=topic, qos=qos, payload="message1", mid=mid, proto_ver=proto_ver)
pubrec1_packet = mosq_test.gen_pubrec(mid=mid, proto_ver=proto_ver)
pubrel1_packet = mosq_test.gen_pubrel(mid=mid, proto_ver=proto_ver)
pubcomp1_packet = mosq_test.gen_pubcomp(mid=mid, proto_ver=proto_ver)
mid = 2
publish2_packet = mosq_test.gen_publish(topic=topic, qos=qos, payload="message2", mid=mid, proto_ver=proto_ver)
pubrec2_packet = mosq_test.gen_pubrec(mid=mid, proto_ver=proto_ver)
pubrel2_packet = mosq_test.gen_pubrel(mid=mid, proto_ver=proto_ver)
pubcomp2_packet = mosq_test.gen_pubcomp(mid=mid, proto_ver=proto_ver)
connect_packet_helper = mosq_test.gen_connect(helper_id, proto_ver=proto_ver, clean_session=True)
subscribe_packet = mosq_test.gen_subscribe(mid, topic, qos=qos, proto_ver=proto_ver)
suback_packet = mosq_test.gen_suback(mid=mid, qos=qos, proto_ver=proto_ver)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
con = None
try:
# Connect client, start flow, disconnect
sock = mosq_test.do_client_connect(connect_packet, connack_packet1, timeout=5, port=port, connack_error="connack 1")
mosq_test.do_send_receive(sock, publish1_packet, pubrec1_packet, "pubrec1 send")
mosq_test.do_send_receive(sock, publish2_packet, pubrec2_packet, "pubrec2 send")
sock.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect helper and subscribe
helper = mosq_test.do_client_connect(connect_packet_helper, connack_packet1, timeout=5, port=port, connack_error="helper connack")
mosq_test.do_send_receive(helper, subscribe_packet, suback_packet, "suback helper")
# Complete the flow
sock = mosq_test.do_client_connect(connect_packet, connack_packet2, timeout=5, port=port, connack_error="connack 2")
mosq_test.do_send_receive(sock, pubrel1_packet, pubcomp1_packet, "pubrel1 send")
mosq_test.do_send_receive(sock, pubrel2_packet, pubcomp2_packet, "pubrel2 send")
mosq_test.expect_packet(helper, "publish1 receive", publish1_packet)
mosq_test.expect_packet(helper, "publish2 receive", publish2_packet)
helper.send(pubrec1_packet)
mosq_test.do_receive_send(helper, pubrel1_packet, pubcomp1_packet, "pubcomp1 receive")
helper.send(pubrec2_packet)
mosq_test.do_receive_send(helper, pubrel2_packet, pubcomp2_packet, "pubcomp2 receive")
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+117
View File
@@ -0,0 +1,117 @@
#!/usr/bin/env python3
from mosq_test_helper import *
import sqlite3
import sqlite_help
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
keepalive = 10
sqlite_help.init(port)
client_id = "sqlite-cmsg-out-dup-v3-1-1"
payload = "queued message 1"
payload_b = payload.encode("UTF-8")
qos = 2
topic = "client-msg/test"
source_id = "sqlite-cmsg-v3-1-1-helper"
proto_ver = 4
keepalive = 10
connect1_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, proto_ver=proto_ver, clean_session=False)
connack1_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
connack1_packet2 = mosq_test.gen_connack(rc=0, proto_ver=proto_ver, flags=1)
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, topic, qos, proto_ver=proto_ver)
suback_packet = mosq_test.gen_suback(mid, qos=qos, proto_ver=proto_ver)
connect2_packet = mosq_test.gen_connect(source_id, keepalive=keepalive, proto_ver=proto_ver)
connack2_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
source_mid = 18
publish_packet = mosq_test.gen_publish(topic, mid=source_mid, qos=qos, payload=payload, proto_ver=proto_ver)
pubrec_packet = mosq_test.gen_pubrec(mid=source_mid, proto_ver=proto_ver)
pubrel_packet = mosq_test.gen_pubrel(mid=source_mid, proto_ver=proto_ver)
pubcomp_packet = mosq_test.gen_pubcomp(mid=source_mid, proto_ver=proto_ver)
mid = 1
publish_packet_r1 = mosq_test.gen_publish(topic, mid=mid, qos=qos, payload=payload, proto_ver=proto_ver)
publish_packet_r2 = mosq_test.gen_publish(topic, mid=mid, qos=qos, payload=payload, proto_ver=proto_ver, dup=1)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
con = None
try:
#con = sqlite3.connect(f"file:{port}/mosquitto.sqlite3?mode=ro", uri=True)
#cur = con.cursor()
# Connect and set up subscription, then disconnect
sock = mosq_test.do_client_connect(connect1_packet, connack1_packet, timeout=5, port=port)
mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
sock.close()
#sqlite_help.check_counts(cur, clients=1, client_msgs=0, messages=0, retains=0, subscriptions=1)
# Helper - send message then disconnect
sock = mosq_test.do_client_connect(connect2_packet, connack2_packet, timeout=5, port=port)
mosq_test.do_send_receive(sock, publish_packet, pubrec_packet, "pubrec")
mosq_test.do_send_receive(sock, pubrel_packet, pubcomp_packet, "pubcomp")
sock.close()
#sqlite_help.check_counts(cur, clients=1, client_msgs=1, messages=1, retains=0, subscriptions=1)
# Reconnect, receive publish, disconnect
sock = mosq_test.do_client_connect(connect1_packet, connack1_packet2, timeout=5, port=port)
mosq_test.expect_packet(sock, "publish 1", publish_packet_r1)
#sqlite_help.check_counts(cur, clients=1, client_msgs=1, messages=1, retains=0, subscriptions=1)
# Reconnect, receive publish, disconnect - dup should now be set
sock = mosq_test.do_client_connect(connect1_packet, connack1_packet2, timeout=5, port=port)
mosq_test.expect_packet(sock, "publish 2", publish_packet_r2)
#con.close()
#con = None
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
broker = None
con = sqlite3.connect(f"{port}/mosquitto.sqlite3")
cur = con.cursor()
sqlite_help.check_counts(cur, clients=1, client_msgs=1, messages=1, retains=0, subscriptions=1)
# Check client
sqlite_help.check_client(cur, client_id, None, 0, 0, port, 0, 2, 1, -1, 0)
# Check subscription
sqlite_help.check_subscription(cur, client_id, topic, qos, 0)
# Check stored message
store_id = sqlite_help.check_store_msg(cur, 0, topic, payload_b, source_id, None, len(payload_b), source_mid, port, qos, 0)
# Check client msg
sqlite_help.check_client_msg(cur, client_id, store_id, 1, sqlite_help.dir_out, 1, qos, 0, sqlite_help.ms_publish_qos2)
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
if con is not None:
con.close()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+88
View File
@@ -0,0 +1,88 @@
#!/usr/bin/env python3
from mosq_test_helper import *
import sqlite3
import sqlite_help
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
keepalive = 10
sqlite_help.init(port)
client_id = "sqlite-cmsg-v3-1-1"
payload = "queued message 1"
payload_b = payload.encode("UTF-8")
qos = 1
topic = "client-msg/test"
source_id = "sqlite-cmsg-v3-1-1-helper"
proto_ver = 4
keepalive = 10
connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, proto_ver=proto_ver, clean_session=False)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, topic, qos, proto_ver=proto_ver)
suback_packet = mosq_test.gen_suback(mid, qos=qos, proto_ver=proto_ver)
connect2_packet = mosq_test.gen_connect(source_id, keepalive=keepalive, proto_ver=proto_ver)
connack2_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
mid = 18
publish_packet = mosq_test.gen_publish(topic, mid=mid, qos=qos, payload=payload, proto_ver=proto_ver)
puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
con = None
try:
sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=5, port=port)
mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
sock.close()
sock = mosq_test.do_client_connect(connect2_packet, connack2_packet, timeout=5, port=port)
mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback")
sock.close()
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
broker = None
con = sqlite3.connect(f"{port}/mosquitto.sqlite3")
cur = con.cursor()
sqlite_help.check_counts(cur, clients=1, client_msgs=1, messages=1, retains=0, subscriptions=1)
# Check client
sqlite_help.check_client(cur, client_id, None, 0, 0, port, 0, 2, 1, -1, 0)
# Check subscription
sqlite_help.check_subscription(cur, client_id, topic, qos, 0)
# Check stored message
store_id = sqlite_help.check_store_msg(cur, 0, topic, payload_b, source_id, None, len(payload_b), mid, port, qos, 0)
# Check client msg
sqlite_help.check_client_msg(cur, client_id, store_id, 0, sqlite_help.dir_out, 1, qos, 0, sqlite_help.ms_queued)
con.close()
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
if con is not None:
con.close()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+92
View File
@@ -0,0 +1,92 @@
#!/usr/bin/env python3
# Connect a client, start a QoS 2 flow, disconnect, restore, carry on with the
# QoS 2 flow. Is it received?
from mosq_test_helper import *
import sqlite_help
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
sqlite_help.init(port)
client_id = "persist-client-msg-in-v5-0"
proto_ver = 5
helper_id = "persist-client-msg-in-v5-0-helper"
topic = "client-msg-in/2"
qos = 2
connect_props = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_SESSION_EXPIRY_INTERVAL, 60)
connect_packet = mosq_test.gen_connect(client_id, proto_ver=proto_ver, clean_session=False, properties=connect_props)
connack_packet1 = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
connack_packet2 = mosq_test.gen_connack(rc=0, flags=1, proto_ver=proto_ver)
mid = 1
publish1_packet = mosq_test.gen_publish(topic=topic, qos=qos, payload="message1", mid=mid, proto_ver=proto_ver)
pubrec1_packet = mosq_test.gen_pubrec(mid=mid, proto_ver=proto_ver)
pubrel1_packet = mosq_test.gen_pubrel(mid=mid, proto_ver=proto_ver)
pubcomp1_packet = mosq_test.gen_pubcomp(mid=mid, proto_ver=proto_ver)
mid = 2
publish2_packet = mosq_test.gen_publish(topic=topic, qos=qos, payload="message2", mid=mid, proto_ver=proto_ver)
pubrec2_packet = mosq_test.gen_pubrec(mid=mid, proto_ver=proto_ver)
pubrel2_packet = mosq_test.gen_pubrel(mid=mid, proto_ver=proto_ver)
pubcomp2_packet = mosq_test.gen_pubcomp(mid=mid, proto_ver=proto_ver)
connect_packet_helper = mosq_test.gen_connect(helper_id, proto_ver=proto_ver, clean_session=True)
subscribe_packet = mosq_test.gen_subscribe(mid, topic, qos=qos, proto_ver=proto_ver)
suback_packet = mosq_test.gen_suback(mid=mid, qos=qos, proto_ver=proto_ver)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
con = None
try:
# Connect client, start flow, disconnect
sock = mosq_test.do_client_connect(connect_packet, connack_packet1, timeout=5, port=port, connack_error="connack 1")
mosq_test.do_send_receive(sock, publish1_packet, pubrec1_packet, "pubrec1 send")
mosq_test.do_send_receive(sock, publish2_packet, pubrec2_packet, "pubrec2 send")
sock.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect helper and subscribe
helper = mosq_test.do_client_connect(connect_packet_helper, connack_packet1, timeout=5, port=port, connack_error="helper connack")
mosq_test.do_send_receive(helper, subscribe_packet, suback_packet, "suback helper")
# Complete the flow
sock = mosq_test.do_client_connect(connect_packet, connack_packet2, timeout=5, port=port, connack_error="connack 2")
mosq_test.do_send_receive(sock, pubrel1_packet, pubcomp1_packet, "pubrel1 send")
mosq_test.do_send_receive(sock, pubrel2_packet, pubcomp2_packet, "pubrel2 send")
mosq_test.expect_packet(helper, "publish1 receive", publish1_packet)
mosq_test.expect_packet(helper, "publish2 receive", publish2_packet)
helper.send(pubrec1_packet)
mosq_test.do_receive_send(helper, pubrel1_packet, pubcomp1_packet, "pubcomp1 receive")
helper.send(pubrec2_packet)
mosq_test.do_receive_send(helper, pubrel2_packet, pubcomp2_packet, "pubcomp2 receive")
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+102
View File
@@ -0,0 +1,102 @@
#!/usr/bin/env python3
# Connect a client, add a subscription, disconnect, send a message with a
# different client, restore, reconnect, check it is received.
from mosq_test_helper import *
import sqlite_help
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
sqlite_help.init(port)
keepalive = 10
client_id = "persist-client-msg-v3-1-1"
proto_ver = 4
helper_id = "persist-client-msg-v3-1-1-helper"
topic0 = "client-msg/0"
topic1 = "client-msg/1"
topic2 = "client-msg/2"
connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, proto_ver=proto_ver, clean_session=False)
connack_packet1 = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
connack_packet2 = mosq_test.gen_connack(rc=0, flags=1, proto_ver=proto_ver)
mid = 1
subscribe_packet0 = mosq_test.gen_subscribe(mid, topic0, qos=0, proto_ver=proto_ver)
suback_packet0 = mosq_test.gen_suback(mid=mid, qos=0, proto_ver=proto_ver)
subscribe_packet1 = mosq_test.gen_subscribe(mid, topic1, qos=1, proto_ver=proto_ver)
suback_packet1 = mosq_test.gen_suback(mid=mid, qos=1, proto_ver=proto_ver)
subscribe_packet2 = mosq_test.gen_subscribe(mid, topic2, qos=2, proto_ver=proto_ver)
suback_packet2 = mosq_test.gen_suback(mid=mid, qos=2, proto_ver=proto_ver)
connect_packet_helper = mosq_test.gen_connect(helper_id, keepalive=keepalive, proto_ver=proto_ver, clean_session=True)
publish_packet0 = mosq_test.gen_publish(topic=topic0, qos=0, payload="message", proto_ver=proto_ver)
mid = 1
publish_packet1 = mosq_test.gen_publish(topic=topic1, qos=1, payload="message", mid=mid, proto_ver=proto_ver)
puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver)
mid = 2
publish_packet2 = mosq_test.gen_publish(topic=topic2, qos=2, payload="message", mid=mid, proto_ver=proto_ver)
pubrec_packet = mosq_test.gen_pubrec(mid=mid, proto_ver=proto_ver)
pubrel_packet = mosq_test.gen_pubrel(mid=mid, proto_ver=proto_ver)
pubcomp_packet = mosq_test.gen_pubcomp(mid=mid, proto_ver=proto_ver)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
con = None
try:
# Connect client, subscribe, disconnect
sock = mosq_test.do_client_connect(connect_packet, connack_packet1, timeout=5, port=port)
mosq_test.do_send_receive(sock, subscribe_packet0, suback_packet0, "suback 0")
mosq_test.do_send_receive(sock, subscribe_packet1, suback_packet1, "suback 1")
mosq_test.do_send_receive(sock, subscribe_packet2, suback_packet2, "suback 2")
sock.close()
# Connect helper and publish
helper = mosq_test.do_client_connect(connect_packet_helper, connack_packet1, timeout=5, port=port)
helper.send(publish_packet0)
mosq_test.do_send_receive(helper, publish_packet1, puback_packet, "puback helper")
mosq_test.do_send_receive(helper, publish_packet2, pubrec_packet, "pubrec helper")
mosq_test.do_send_receive(helper, pubrel_packet, pubcomp_packet, "pubcomp helper")
helper.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect client again, it should have a session
sock = mosq_test.do_client_connect(connect_packet, connack_packet2, timeout=5, port=port)
# Does the client get the messages
mosq_test.do_receive_send(sock, publish_packet1, puback_packet, "publish 1")
mosq_test.do_receive_send(sock, publish_packet2, pubrec_packet, "publish 2")
mosq_test.do_receive_send(sock, pubrel_packet, pubcomp_packet, "pubrel 2")
sock.close()
# Connect client again, it should have a session
sock = mosq_test.do_client_connect(connect_packet, connack_packet2, timeout=5, port=port)
# If there are messages, the ping will fail
mosq_test.do_ping(sock)
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+102
View File
@@ -0,0 +1,102 @@
#!/usr/bin/env python3
# Connect a client, add a subscription, disconnect, send a message with a
# different client, restore, reconnect, check it is received.
from mosq_test_helper import *
import sqlite_help
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
sqlite_help.init(port)
keepalive = 10
client_id = "persist-client-msg-v3-1-1"
proto_ver = 4
helper_id = "persist-client-msg-v3-1-1-helper"
topic0 = "client-msg/0"
topic1 = "client-msg/1"
topic2 = "client-msg/2"
connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, proto_ver=proto_ver, clean_session=False)
connack_packet1 = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
connack_packet2 = mosq_test.gen_connack(rc=0, flags=1, proto_ver=proto_ver)
mid = 1
subscribe_packet0 = mosq_test.gen_subscribe(mid, topic0, qos=0, proto_ver=proto_ver)
suback_packet0 = mosq_test.gen_suback(mid=mid, qos=0, proto_ver=proto_ver)
subscribe_packet1 = mosq_test.gen_subscribe(mid, topic1, qos=1, proto_ver=proto_ver)
suback_packet1 = mosq_test.gen_suback(mid=mid, qos=1, proto_ver=proto_ver)
subscribe_packet2 = mosq_test.gen_subscribe(mid, topic2, qos=2, proto_ver=proto_ver)
suback_packet2 = mosq_test.gen_suback(mid=mid, qos=2, proto_ver=proto_ver)
connect_packet_helper = mosq_test.gen_connect(helper_id, keepalive=keepalive, proto_ver=proto_ver, clean_session=True)
publish_packet0 = mosq_test.gen_publish(topic=topic0, qos=0, payload="message", proto_ver=proto_ver)
mid = 1
publish_packet1 = mosq_test.gen_publish(topic=topic1, qos=1, payload="message", mid=mid, proto_ver=proto_ver)
puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver)
mid = 2
publish_packet2 = mosq_test.gen_publish(topic=topic2, qos=2, payload="message", mid=mid, proto_ver=proto_ver)
pubrec_packet = mosq_test.gen_pubrec(mid=mid, proto_ver=proto_ver)
pubrel_packet = mosq_test.gen_pubrel(mid=mid, proto_ver=proto_ver)
pubcomp_packet = mosq_test.gen_pubcomp(mid=mid, proto_ver=proto_ver)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
con = None
try:
# Connect client, subscribe, disconnect
sock = mosq_test.do_client_connect(connect_packet, connack_packet1, timeout=5, port=port)
mosq_test.do_send_receive(sock, subscribe_packet0, suback_packet0, "suback 0")
mosq_test.do_send_receive(sock, subscribe_packet1, suback_packet1, "suback 1")
mosq_test.do_send_receive(sock, subscribe_packet2, suback_packet2, "suback 2")
sock.close()
# Connect helper and publish
helper = mosq_test.do_client_connect(connect_packet_helper, connack_packet1, timeout=5, port=port)
helper.send(publish_packet0)
mosq_test.do_send_receive(helper, publish_packet1, puback_packet, "puback helper")
mosq_test.do_send_receive(helper, publish_packet2, pubrec_packet, "pubrec helper")
mosq_test.do_send_receive(helper, pubrel_packet, pubcomp_packet, "pubcomp helper")
helper.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect client again, it should have a session
sock = mosq_test.do_client_connect(connect_packet, connack_packet2, timeout=5, port=port)
# Does the client get the messages
mosq_test.do_receive_send(sock, publish_packet1, puback_packet, "publish 1")
mosq_test.do_receive_send(sock, publish_packet2, pubrec_packet, "publish 2")
mosq_test.do_receive_send(sock, pubrel_packet, pubcomp_packet, "pubrel 2")
sock.close()
# Connect client again, it should have a session
sock = mosq_test.do_client_connect(connect_packet, connack_packet2, timeout=5, port=port)
# If there are messages, the ping will fail
mosq_test.do_ping(sock)
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+102
View File
@@ -0,0 +1,102 @@
#!/usr/bin/env python3
# Connect a client, add a subscription, disconnect, send a message with a
# different client, restore, reconnect, check it is received.
from mosq_test_helper import *
import sqlite_help
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
sqlite_help.init(port)
client_id = "persist-client-msg-v5-0"
proto_ver = 5
helper_id = "persist-client-msg-v5-0-helper"
topic0 = "client-msg/0"
topic1 = "client-msg/1"
topic2 = "client-msg/2"
connect_props = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_SESSION_EXPIRY_INTERVAL, 60)
connect_packet = mosq_test.gen_connect(client_id, proto_ver=proto_ver, clean_session=False, properties=connect_props)
connack_packet1 = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
connack_packet2 = mosq_test.gen_connack(rc=0, flags=1, proto_ver=proto_ver)
mid = 1
subscribe_packet0 = mosq_test.gen_subscribe(mid, topic0, qos=0, proto_ver=proto_ver)
suback_packet0 = mosq_test.gen_suback(mid=mid, qos=0, proto_ver=proto_ver)
subscribe_packet1 = mosq_test.gen_subscribe(mid, topic1, qos=1, proto_ver=proto_ver)
suback_packet1 = mosq_test.gen_suback(mid=mid, qos=1, proto_ver=proto_ver)
subscribe_packet2 = mosq_test.gen_subscribe(mid, topic2, qos=2, proto_ver=proto_ver)
suback_packet2 = mosq_test.gen_suback(mid=mid, qos=2, proto_ver=proto_ver)
connect_packet_helper = mosq_test.gen_connect(helper_id, proto_ver=proto_ver, clean_session=True)
publish_packet0 = mosq_test.gen_publish(topic=topic0, qos=0, payload="message", proto_ver=proto_ver)
mid = 1
publish_packet1 = mosq_test.gen_publish(topic=topic1, qos=1, payload="message", mid=mid, proto_ver=proto_ver)
puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver)
mid = 2
publish_packet2 = mosq_test.gen_publish(topic=topic2, qos=2, payload="message", mid=mid, proto_ver=proto_ver)
pubrec_packet = mosq_test.gen_pubrec(mid=mid, proto_ver=proto_ver)
pubrel_packet = mosq_test.gen_pubrel(mid=mid, proto_ver=proto_ver)
pubcomp_packet = mosq_test.gen_pubcomp(mid=mid, proto_ver=proto_ver)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
con = None
try:
# Connect client, subscribe, disconnect
sock = mosq_test.do_client_connect(connect_packet, connack_packet1, timeout=5, port=port, connack_error="connack 1")
mosq_test.do_send_receive(sock, subscribe_packet0, suback_packet0, "suback 0")
mosq_test.do_send_receive(sock, subscribe_packet1, suback_packet1, "suback 1")
mosq_test.do_send_receive(sock, subscribe_packet2, suback_packet2, "suback 2")
sock.close()
# Connect helper and publish
helper = mosq_test.do_client_connect(connect_packet_helper, connack_packet1, timeout=5, port=port, connack_error="helper connack")
helper.send(publish_packet0)
mosq_test.do_send_receive(helper, publish_packet1, puback_packet, "puback helper")
mosq_test.do_send_receive(helper, publish_packet2, pubrec_packet, "pubrec helper")
mosq_test.do_send_receive(helper, pubrel_packet, pubcomp_packet, "pubcomp helper")
helper.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect client again, it should have a session
sock = mosq_test.do_client_connect(connect_packet, connack_packet2, timeout=5, port=port, connack_error="connack 2")
# Does the client get the messages
mosq_test.do_receive_send(sock, publish_packet1, puback_packet, "publish 1")
mosq_test.do_receive_send(sock, publish_packet2, pubrec_packet, "publish 2")
mosq_test.do_receive_send(sock, pubrel_packet, pubcomp_packet, "pubrel 2")
sock.close()
# Connect client again, it should have a session
sock = mosq_test.do_client_connect(connect_packet, connack_packet2, timeout=5, port=port, connack_error="connack 3")
# If there are messages, the ping will fail
mosq_test.do_ping(sock)
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+87
View File
@@ -0,0 +1,87 @@
#!/usr/bin/env python3
# Connect a client, check it is restored, clear the client, check it is not there.
from mosq_test_helper import *
import sqlite_help
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
sqlite_help.init(port)
keepalive = 10
client_id = "persist-client-v3-1-1"
proto_ver = 4
connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, proto_ver=proto_ver, clean_session=False)
connack_packet1 = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
connack_packet2 = mosq_test.gen_connack(rc=0, flags=1, proto_ver=proto_ver)
connect_packet_clean = mosq_test.gen_connect(client_id, keepalive=keepalive, proto_ver=proto_ver, clean_session=True)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
con = None
try:
# Connect client
sock = mosq_test.do_client_connect(connect_packet, connack_packet1, timeout=5, port=port, connack_error="connack 1")
mosq_test.do_ping(sock)
sock.close()
# Connect client again, it should have a session
sock = mosq_test.do_client_connect(connect_packet, connack_packet2, timeout=5, port=port, connack_error="connack 2")
mosq_test.do_ping(sock)
sock.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect client again, it should have a session
sock = mosq_test.do_client_connect(connect_packet, connack_packet2, timeout=5, port=port, connack_error="connack 3")
mosq_test.do_ping(sock)
sock.close()
# Clear the client
sock = mosq_test.do_client_connect(connect_packet_clean, connack_packet1, timeout=5, port=port, connack_error="connack 4")
mosq_test.do_ping(sock)
sock.close()
# Connect client, it should not have a session
sock = mosq_test.do_client_connect(connect_packet_clean, connack_packet1, timeout=5, port=port, connack_error="connack 5")
mosq_test.do_ping(sock)
sock.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect client, it should not have a session
sock = mosq_test.do_client_connect(connect_packet_clean, connack_packet1, timeout=5, port=port, connack_error="connack 6")
mosq_test.do_ping(sock)
sock.close()
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+69
View File
@@ -0,0 +1,69 @@
#!/usr/bin/env python3
# Connect a single client with session expiry interval > 0 and check the
# persisted DB is correct
from mosq_test_helper import *
import sqlite3
import sqlite_help
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
keepalive = 10
sqlite_help.init(port)
keepalive = 10
props = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_SESSION_EXPIRY_INTERVAL, 60)
props += mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_MAXIMUM_PACKET_SIZE, 10000)
connect_packet = mosq_test.gen_connect("sqlite-client-v5-0", keepalive=keepalive, proto_ver=5, properties=props)
props = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_TOPIC_ALIAS_MAXIMUM, 10)
props += mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 20)
#props += mqtt5_props.gen_byte_prop(mqtt5_props.PROP_MAXIMUM_QOS, 1)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props, property_helper=False)
connect_packet_clean = mosq_test.gen_connect("sqlite-client-v5-0-clean", keepalive=keepalive, proto_ver=5)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
con = None
try:
sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=5, port=port, connack_error="connack 1")
sock.close()
sock = mosq_test.do_client_connect(connect_packet_clean, connack_packet, timeout=5, port=port, connack_error="connack 2")
sock.close()
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
broker = None
# Verify sqlite db
con = sqlite3.connect(f"{port}/mosquitto.sqlite3")
cur = con.cursor()
sqlite_help.check_counts(cur, clients=1, client_msgs=0, messages=0, retains=0, subscriptions=0)
# Check client
sqlite_help.check_client(cur, "sqlite-client-v5-0", None, 0, 1, port, 10000, 2, 1, 60, 0)
con.close()
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
if con is not None:
con.close()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+83
View File
@@ -0,0 +1,83 @@
#!/usr/bin/env python3
# Connect a client, check it is restored, clear the client, check it is not there.
from mosq_test_helper import *
import sqlite_help
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
sqlite_help.init(port)
keepalive = 10
client_id = "persist-client-v5-0"
proto_ver = 5
connect_props = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_SESSION_EXPIRY_INTERVAL, 60)
connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, proto_ver=proto_ver, clean_session=False, properties=connect_props)
connack_packet1 = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
connack_packet2 = mosq_test.gen_connack(rc=0, flags=1, proto_ver=proto_ver)
connect_packet_clean = mosq_test.gen_connect(client_id, keepalive=keepalive, proto_ver=proto_ver, clean_session=True)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
con = None
try:
# Connect client
sock = mosq_test.do_client_connect(connect_packet, connack_packet1, timeout=5, port=port, connack_error="connack 1")
mosq_test.do_ping(sock)
sock.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect client again, it should have a session
sock = mosq_test.do_client_connect(connect_packet, connack_packet2, timeout=5, port=port, connack_error="connack 2")
mosq_test.do_ping(sock)
sock.close()
# Clear the client
sock = mosq_test.do_client_connect(connect_packet_clean, connack_packet1, timeout=5, port=port, connack_error="connack 3")
mosq_test.do_ping(sock)
sock.close()
# Connect client, it should not have a session
sock = mosq_test.do_client_connect(connect_packet_clean, connack_packet1, timeout=5, port=port, connack_error="connack 4")
mosq_test.do_ping(sock)
sock.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect client, it should not have a session
sock = mosq_test.do_client_connect(connect_packet_clean, connack_packet1, timeout=5, port=port, connack_error="connack 5")
mosq_test.do_ping(sock)
sock.close()
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+86
View File
@@ -0,0 +1,86 @@
#!/usr/bin/env python3
# Publish a retained messages, check they are restored, with properties attached
from mosq_test_helper import *
import sqlite_help
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
sqlite_help.init(port)
topic = "test/retainprop"
source_id = "persist-retain-properties-v5-0"
qos = 0
proto_ver = 5
connect_packet = mosq_test.gen_connect(source_id, proto_ver=proto_ver, clean_session=True)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
props = mqtt5_props.gen_byte_prop(mqtt5_props.PROP_PAYLOAD_FORMAT_INDICATOR, 1)
props += mqtt5_props.gen_string_prop(mqtt5_props.PROP_CONTENT_TYPE, "plain/text")
props += mqtt5_props.gen_string_prop(mqtt5_props.PROP_RESPONSE_TOPIC, "/dev/null")
#props += mqtt5_props.gen_string_prop(mqtt5_props.PROP_CORRELATION_DATA, "2357289375902345")
props += mqtt5_props.gen_string_pair_prop(mqtt5_props.PROP_USER_PROPERTY, "name", "value")
props += mqtt5_props.gen_string_pair_prop(mqtt5_props.PROP_USER_PROPERTY, "name", "value")
props += mqtt5_props.gen_string_pair_prop(mqtt5_props.PROP_USER_PROPERTY, "name", "value")
props += mqtt5_props.gen_string_pair_prop(mqtt5_props.PROP_USER_PROPERTY, "name", "value")
publish_packet = mosq_test.gen_publish(topic, qos=qos, payload="retained message 1", retain=True, proto_ver=proto_ver, properties=props)
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "test/retainprop", 0, proto_ver=proto_ver)
suback_packet = mosq_test.gen_suback(mid, qos=0, proto_ver=proto_ver)
mid = 2
unsubscribe_packet = mosq_test.gen_unsubscribe(mid, "test/retainprop", proto_ver=proto_ver)
unsuback_packet = mosq_test.gen_unsuback(mid, proto_ver=proto_ver)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
try:
# Connect client
sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=5, port=port)
# Check no retained messages exist
mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
# Ping will fail if a PUBLISH is received
mosq_test.do_ping(sock)
# Unsubscribe, so we don't receive the messages
mosq_test.do_send_receive(sock, unsubscribe_packet, unsuback_packet, "unsuback")
# Send some retained messages
sock.send(publish_packet)
mosq_test.do_ping(sock)
sock.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect client
sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=5, port=port)
# Subscribe
mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
# Check retained messages exist
mosq_test.expect_packet(sock, "publish", publish_packet)
mosq_test.do_ping(sock)
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+89
View File
@@ -0,0 +1,89 @@
#!/usr/bin/env python3
# Publish a retained messages, check they are restored
from mosq_test_helper import *
import sqlite_help
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
sqlite_help.init(port)
keepalive = 10
topic1 = "test/retain1"
topic2 = "test/retain2"
topic3 = "test/retain3"
source_id = "persist-retain-v3-1-1"
qos = 0
payload2 = "retained message 2"
payload3 = "retained message 3"
proto_ver = 4
connect_packet = mosq_test.gen_connect(source_id, keepalive=keepalive, proto_ver=proto_ver, clean_session=True)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
publish1_packet = mosq_test.gen_publish(topic1, qos=qos, payload="retained message 1", retain=True, proto_ver=proto_ver)
publish2_packet = mosq_test.gen_publish(topic2, qos=qos, payload=payload2, retain=False, proto_ver=proto_ver)
publish3_packet = mosq_test.gen_publish(topic3, qos=qos, payload=payload3, retain=True, proto_ver=proto_ver)
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "#", 0, proto_ver=4)
suback_packet = mosq_test.gen_suback(mid, qos=0, proto_ver=4)
mid = 2
unsubscribe_packet = mosq_test.gen_unsubscribe(mid, "#", proto_ver=4)
unsuback_packet = mosq_test.gen_unsuback(mid, proto_ver=4)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
try:
# Connect client
sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=5, port=port)
# Check no retained messages exist
mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
# Ping will fail if a PUBLISH is received
mosq_test.do_ping(sock)
# Unsubscribe, so we don't receive the messages
mosq_test.do_send_receive(sock, unsubscribe_packet, unsuback_packet, "unsuback")
# Send some retained messages
sock.send(publish1_packet)
mosq_test.do_ping(sock)
sock.send(publish2_packet) # Not retained
mosq_test.do_ping(sock)
sock.send(publish3_packet)
mosq_test.do_ping(sock)
sock.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect client
sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=5, port=port)
# Subscribe
mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
# Check retained messages exist
mosq_test.receive_unordered(sock, publish1_packet, publish3_packet, "publish 1 / 3")
mosq_test.do_ping(sock)
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+88
View File
@@ -0,0 +1,88 @@
#!/usr/bin/env python3
# Publish a retained messages, check they are restored
from mosq_test_helper import *
import sqlite_help
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
sqlite_help.init(port)
topic1 = "test/retain1"
topic2 = "test/retain2"
topic3 = "test/retain3"
source_id = "persist-retain-v5-0"
qos = 0
payload2 = "retained message 2"
payload3 = "retained message 3"
proto_ver = 5
connect_packet = mosq_test.gen_connect(source_id, proto_ver=proto_ver, clean_session=True)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
publish1_packet = mosq_test.gen_publish(topic1, qos=qos, payload="retained message 1", retain=True, proto_ver=proto_ver)
publish2_packet = mosq_test.gen_publish(topic2, qos=qos, payload=payload2, retain=False, proto_ver=proto_ver)
publish3_packet = mosq_test.gen_publish(topic3, qos=qos, payload=payload3, retain=True, proto_ver=proto_ver)
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "#", 0, proto_ver=proto_ver)
suback_packet = mosq_test.gen_suback(mid, qos=0, proto_ver=proto_ver)
mid = 2
unsubscribe_packet = mosq_test.gen_unsubscribe(mid, "#", proto_ver=proto_ver)
unsuback_packet = mosq_test.gen_unsuback(mid, proto_ver=proto_ver)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
try:
# Connect client
sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=5, port=port)
# Check no retained messages exist
mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
# Ping will fail if a PUBLISH is received
mosq_test.do_ping(sock)
# Unsubscribe, so we don't receive the messages
mosq_test.do_send_receive(sock, unsubscribe_packet, unsuback_packet, "unsuback")
# Send some retained messages
sock.send(publish1_packet)
mosq_test.do_ping(sock)
sock.send(publish2_packet) # Not retained
mosq_test.do_ping(sock)
sock.send(publish3_packet)
mosq_test.do_ping(sock)
sock.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect client
sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=5, port=port)
# Subscribe
mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
# Check retained messages exist
mosq_test.receive_unordered(sock, publish1_packet, publish3_packet, "publish 1 / 3")
mosq_test.do_ping(sock)
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+128
View File
@@ -0,0 +1,128 @@
#!/usr/bin/env python3
# Connect a client, add a subscription, disconnect, restore, reconnect, send a
# message with a different client, check it is received.
from mosq_test_helper import *
import sqlite_help
def helper(port, packets):
helper_id = "persist-subscription-v3-1-1-helper"
connect_packet_helper = mosq_test.gen_connect(helper_id, proto_ver=4, clean_session=True)
# Connect helper and publish
helper = mosq_test.do_client_connect(connect_packet_helper, packets["connack1"], timeout=5, port=port)
helper.send(packets["publish0"])
mosq_test.do_send_receive(helper, packets["publish1"], packets["puback1"], "puback helper")
mosq_test.do_send_receive(helper, packets["publish2"], packets["pubrec2"], "pubrec helper")
mosq_test.do_send_receive(helper, packets["pubrel2"], packets["pubcomp2"], "pubcomp helper")
helper.close()
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
sqlite_help.init(port)
client_id = "persist-subscription-v3-1-1"
proto_ver = 4
topic0 = "subscription/0"
topic1 = "subscription/1"
topic2 = "subscription/2"
packets = {}
packets["connect"] = mosq_test.gen_connect(client_id, proto_ver=proto_ver, clean_session=False)
packets["connack1"] = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
packets["connack2"] = mosq_test.gen_connack(rc=0, flags=1, proto_ver=proto_ver)
mid = 1
packets["subscribe0"] = mosq_test.gen_subscribe(mid, topic0, qos=0, proto_ver=proto_ver)
packets["suback0"] = mosq_test.gen_suback(mid=mid, qos=0, proto_ver=proto_ver)
packets["subscribe1"] = mosq_test.gen_subscribe(mid, topic1, qos=1, proto_ver=proto_ver)
packets["suback1"] = mosq_test.gen_suback(mid=mid, qos=1, proto_ver=proto_ver)
packets["subscribe2"] = mosq_test.gen_subscribe(mid, topic2, qos=2, proto_ver=proto_ver)
packets["suback2"] = mosq_test.gen_suback(mid=mid, qos=2, proto_ver=proto_ver)
packets["unsubscribe2"] = mosq_test.gen_unsubscribe(mid, topic2, proto_ver=proto_ver)
packets["unsuback2"] = mosq_test.gen_unsuback(mid=mid, proto_ver=proto_ver)
packets["publish0"] = mosq_test.gen_publish(topic=topic0, qos=0, payload="message", proto_ver=proto_ver)
mid = 1
packets["publish1"] = mosq_test.gen_publish(topic=topic1, qos=1, payload="message", mid=mid, proto_ver=proto_ver)
packets["puback1"] = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver)
mid = 2
packets["publish2"] = mosq_test.gen_publish(topic=topic2, qos=2, payload="message", mid=mid, proto_ver=proto_ver)
packets["pubrec2"] = mosq_test.gen_pubrec(mid=mid, proto_ver=proto_ver)
packets["pubrel2"] = mosq_test.gen_pubrel(mid=mid, proto_ver=proto_ver)
packets["pubcomp2"] = mosq_test.gen_pubcomp(mid=mid, proto_ver=proto_ver)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
con = None
try:
# Connect client
sock = mosq_test.do_client_connect(packets["connect"], packets["connack1"], timeout=5, port=port)
mosq_test.do_send_receive(sock, packets["subscribe0"], packets["suback0"], "suback 0")
mosq_test.do_send_receive(sock, packets["subscribe1"], packets["suback1"], "suback 1")
mosq_test.do_send_receive(sock, packets["subscribe2"], packets["suback2"], "suback 2")
sock.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect client again, it should have a session
sock = mosq_test.do_client_connect(packets["connect"], packets["connack2"], timeout=5, port=port)
mosq_test.do_ping(sock)
helper(port, packets)
# Does the client get the messages
mosq_test.expect_packet(sock, "publish 0", packets["publish0"])
mosq_test.do_receive_send(sock, packets["publish1"], packets["puback1"], "publish 1")
mosq_test.do_receive_send(sock, packets["publish2"], packets["pubrec2"], "publish 2")
mosq_test.do_receive_send(sock, packets["pubrel2"], packets["pubcomp2"], "pubrel 2")
# Unsubscribe
mosq_test.do_send_receive(sock, packets["unsubscribe2"], packets["unsuback2"], "unsuback 2")
sock.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect client again, it should have a session
sock = mosq_test.do_client_connect(packets["connect"], packets["connack2"], timeout=5, port=port)
mosq_test.do_ping(sock)
# Connect helper and publish
helper(port, packets)
# Does the client get the messages
mosq_test.expect_packet(sock, "publish 0", packets["publish0"])
mosq_test.do_receive_send(sock, packets["publish1"], packets["puback1"], "publish 1")
mosq_test.do_ping(sock)
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+139
View File
@@ -0,0 +1,139 @@
#!/usr/bin/env python3
# Connect a client, add a subscription, disconnect, restore, reconnect, send a
# message with a different client, check it is received.
from mosq_test_helper import *
import sqlite_help
def helper(port, packets):
helper_id = "persist-subscription-v5-0-helper"
connect_packet_helper = mosq_test.gen_connect(helper_id, proto_ver=5, clean_session=True)
connack_packet_helper = mosq_test.gen_connack(rc=0, proto_ver=5)
# Connect helper and publish
helper = mosq_test.do_client_connect(connect_packet_helper, connack_packet_helper, timeout=5, port=port, connack_error="helper connack")
helper.send(packets["publish0-helper"])
mosq_test.do_send_receive(helper, packets["publish1-helper"], packets["puback1"], "puback helper")
mosq_test.do_send_receive(helper, packets["publish2-helper"], packets["pubrec2"], "pubrec helper")
mosq_test.do_send_receive(helper, packets["pubrel2"], packets["pubcomp2"], "pubcomp helper")
helper.close()
port = mosq_test.get_port()
conf_file = os.path.basename(__file__).replace('.py', '.conf')
sqlite_help.write_config(conf_file, port)
rc = 1
sqlite_help.init(port)
client_id = "persist-subscription-v5-0"
proto_ver = 5
topic0 = "subscription/0"
topic1 = "subscription/1"
topic2 = "subscription/2"
packets = {}
connect_props = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_SESSION_EXPIRY_INTERVAL, 60)
packets["connect"] = mosq_test.gen_connect(client_id, proto_ver=proto_ver, clean_session=False, properties=connect_props)
packets["connack1"] = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
packets["connack2"] = mosq_test.gen_connack(rc=0, flags=1, proto_ver=proto_ver)
mid = 1
publish_props0 = mqtt5_props.gen_varint_prop(mqtt5_props.PROP_SUBSCRIPTION_IDENTIFIER, 100)
packets["subscribe0"] = mosq_test.gen_subscribe(mid, topic0, qos=0, proto_ver=proto_ver, properties=publish_props0)
packets["suback0"] = mosq_test.gen_suback(mid=mid, qos=0, proto_ver=proto_ver)
publish_props1 = mqtt5_props.gen_varint_prop(mqtt5_props.PROP_SUBSCRIPTION_IDENTIFIER, 101)
packets["subscribe1"] = mosq_test.gen_subscribe(mid, topic1, qos=1, proto_ver=proto_ver, properties=publish_props1)
packets["suback1"] = mosq_test.gen_suback(mid=mid, qos=1, proto_ver=proto_ver)
publish_props2 = mqtt5_props.gen_varint_prop(mqtt5_props.PROP_SUBSCRIPTION_IDENTIFIER, 102)
packets["subscribe2"] = mosq_test.gen_subscribe(mid, topic2, qos=2, proto_ver=proto_ver, properties=publish_props2)
packets["suback2"] = mosq_test.gen_suback(mid=mid, qos=2, proto_ver=proto_ver)
packets["unsubscribe2"] = mosq_test.gen_unsubscribe(mid, topic2, proto_ver=proto_ver)
packets["unsuback2"] = mosq_test.gen_unsuback(mid=mid, proto_ver=proto_ver)
packets["publish0-helper"] = mosq_test.gen_publish(topic=topic0, qos=0, payload="message", proto_ver=proto_ver)
packets["publish0"] = mosq_test.gen_publish(topic=topic0, qos=0, payload="message", proto_ver=proto_ver, properties=publish_props0)
mid = 1
packets["publish1-helper"] = mosq_test.gen_publish(topic=topic1, qos=1, payload="message", mid=mid, proto_ver=proto_ver)
packets["publish1"] = mosq_test.gen_publish(topic=topic1, qos=1, payload="message", mid=mid, proto_ver=proto_ver, properties=publish_props1)
packets["puback1"] = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver)
mid = 2
packets["publish2-helper"] = mosq_test.gen_publish(topic=topic2, qos=2, payload="message", mid=mid, proto_ver=proto_ver)
packets["publish2"] = mosq_test.gen_publish(topic=topic2, qos=2, payload="message", mid=mid, proto_ver=proto_ver, properties=publish_props2)
packets["pubrec2"] = mosq_test.gen_pubrec(mid=mid, proto_ver=proto_ver)
packets["pubrel2"] = mosq_test.gen_pubrel(mid=mid, proto_ver=proto_ver)
packets["pubcomp2"] = mosq_test.gen_pubcomp(mid=mid, proto_ver=proto_ver)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
con = None
try:
# Connect client
sock = mosq_test.do_client_connect(packets["connect"], packets["connack1"], timeout=5, port=port, connack_error="connack1")
mosq_test.do_send_receive(sock, packets["subscribe0"], packets["suback0"], "suback 0")
mosq_test.do_send_receive(sock, packets["subscribe1"], packets["suback1"], "suback 1")
mosq_test.do_send_receive(sock, packets["subscribe2"], packets["suback2"], "suback 2")
sock.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect client again, it should have a session
sock = mosq_test.do_client_connect(packets["connect"], packets["connack2"], timeout=5, port=port)
mosq_test.do_ping(sock)
helper(port, packets)
# Does the client get the messages
mosq_test.expect_packet(sock, "publish 0", packets["publish0"])
mosq_test.do_receive_send(sock, packets["publish1"], packets["puback1"], "publish 1")
mosq_test.do_receive_send(sock, packets["publish2"], packets["pubrec2"], "publish 2")
mosq_test.do_receive_send(sock, packets["pubrel2"], packets["pubcomp2"], "pubrel 2")
# Unsubscribe
mosq_test.do_send_receive(sock, packets["unsubscribe2"], packets["unsuback2"], "unsuback 2")
sock.close()
# Kill broker
broker.terminate()
broker.wait()
# Restart broker
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
# Connect client again, it should have a session
sock = mosq_test.do_client_connect(packets["connect"], packets["connack2"], timeout=5, port=port, connack_error="connack2")
mosq_test.do_ping(sock)
# Connect helper and publish
helper(port, packets)
# Does the client get the messages
mosq_test.expect_packet(sock, "publish 0", packets["publish0"])
mosq_test.do_receive_send(sock, packets["publish1"], packets["puback1"], "publish 1")
mosq_test.do_ping(sock)
rc = 0
finally:
if broker is not None:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
os.remove(conf_file)
rc += sqlite_help.cleanup(port)
if rc:
print(stde.decode('utf-8'))
exit(rc)
+15
View File
@@ -264,6 +264,21 @@ endif
#./15-persist-retain-v5-0.py
#./15-persist-subscription-v3-1-1.py
#./15-persist-subscription-v5-0.py
./15-sqlite-client-message-in-v3-1-1.py
./15-sqlite-client-message-out-dup-v3-1-1.py
./15-sqlite-client-message-out-v3-1-1.py
./15-sqlite-client-msg-in-v5-0.py
./15-sqlite-client-msg-out-queue-v3-1-1.py
./15-sqlite-client-msg-out-v3-1-1.py
./15-sqlite-client-msg-out-v5-0.py
./15-sqlite-client-v3-1-1.py
./15-sqlite-client-v5-0.py
./15-sqlite-client-v5.0.py
./15-sqlite-publish-properties-v5-0.py
./15-sqlite-retain-v3-1-1.py
./15-sqlite-retain-v5-0.py
./15-sqlite-subscription-v3-1-1.py
./15-sqlite-subscription-v5-0.py
16 :
./16-cmd-args.py
+217
View File
@@ -0,0 +1,217 @@
import os
dir_in = 0
dir_out = 1
ms_invalid = 0
ms_publish_qos0 = 1
ms_publish_qos1 = 2
ms_wait_for_puback = 3
ms_publish_qos2 = 4
ms_wait_for_pubrec = 5
ms_resend_pubrel = 6
ms_wait_for_pubrel = 7
ms_resend_pubcomp = 8
ms_wait_for_pubcomp = 9
ms_send_pubrec = 10
ms_queued = 11
def write_config(filename, port):
with open(filename, 'w') as f:
f.write("listener %d\n" % (port))
f.write("allow_anonymous true\n")
f.write("plugin ../../plugins/persist-sqlite/mosquitto_persist_sqlite.so\n")
f.write("plugin_opt_db_file %d/mosquitto.sqlite3\n" % (port))
def init(port):
try:
os.mkdir(str(port))
except FileExistsError:
pass
def cleanup(port):
rc = 1
try:
os.remove(f"{port}/mosquitto.sqlite3")
except FileNotFoundError:
pass
try:
os.rmdir(f"{port}")
rc = 0
except OSError:
os.remove(f"{port}/mosquitto.sqlite3-shm")
os.remove(f"{port}/mosquitto.sqlite3-wal")
os.rmdir(f"{port}")
return rc
def check_counts(cur, clients, client_msgs, messages, retains, subscriptions):
cur.execute('SELECT COUNT(*) FROM clients')
row = cur.fetchone()
if row[0] != clients:
raise ValueError("Found %d clients, expected %d" % (row[0], clients))
cur.execute('SELECT COUNT(*) FROM client_msgs')
row = cur.fetchone()
if row[0] != client_msgs:
raise ValueError("Found %d client_msgs, expected %d" % (row[0], client_msgs))
cur.execute('SELECT COUNT(*) FROM subscriptions')
row = cur.fetchone()
if row[0] != subscriptions:
raise ValueError("Found %d subscriptions, expected %d" % (row[0], subscriptions))
cur.execute('SELECT COUNT(*) FROM msgs')
row = cur.fetchone()
if row[0] != messages:
raise ValueError("Found %d msgs, expected %d" % (row[0], messages))
cur.execute('SELECT COUNT(*) FROM retains')
row = cur.fetchone()
if row[0] != retains:
raise ValueError("Found %d retains, expected %d" % (row[0], retains))
def check_client(cur, client_id, username, will_delay_time, session_expiry_time,
listener_port, max_packet_size, max_qos, retain_available,
session_expiry_interval, will_delay_interval):
cur.execute('SELECT client_id, username, will_delay_time, session_expiry_time, ' +
'listener_port, max_packet_size, max_qos, retain_available, ' +
'session_expiry_interval, will_delay_interval ' +
'FROM clients')
row = cur.fetchone()
if row[0] != client_id:
raise ValueError("Invalid client_id %s / %s" % (row[0], client_id))
if row[1] != username:
raise ValueError("Invalid username %s / %s" % (row[1], username))
if (will_delay_time == 0 and row[2] != 0) or (will_delay_time != 0 and row[2] == 0):
raise ValueError("Invalid will_delay_time %d / %d" % (row[2], will_delay_time))
if (session_expiry_time == 0 and row[3] != 0) or (session_expiry_time != 0 and row[3] == 0):
raise ValueError("Invalid session_expiry_time %d / %d" % (row[3], session_expiry_time))
if row[4] != listener_port:
raise ValueError("Invalid listener_port %d / %d" % (row[4], listener_port))
if row[5] != max_packet_size:
raise ValueError("Invalid max_packet_size %d / %d" % (row[5], max_packet_size))
if row[6] != max_qos:
raise ValueError("Invalid max_qos %d / %d" % (row[6], max_qos))
if row[7] != retain_available:
raise ValueError("Invalid retain_available %d / %d" % (row[7], retain_available))
if row[8] != session_expiry_interval:
raise ValueError("Invalid session_expiry_interval %d / %d" % (row[8], session_expiry_interval))
if row[9] != will_delay_interval:
raise ValueError("Invalid will_delay_interval %d / %d" % (row[9], will_delay_interval))
def check_subscription(cur, client_id, topic, subscription_options, subscription_identifier):
cur.execute('SELECT client_id, topic, subscription_options, subscription_identifier ' +
'FROM subscriptions')
row = cur.fetchone()
if row[0] != client_id:
raise ValueError("Invalid client_id %s / %s" % (row[0], client_id))
if row[1] != topic:
raise ValueError("Invalid topic %s / %s" % (row[1], topic))
if row[2] != subscription_options:
raise ValueError("Invalid subscription_options %d / %d" % (row[2], subscription_options))
if row[3] != subscription_identifier:
raise ValueError("Invalid subscription_identifier %d / %d" % (row[3], subscription_identifier))
def check_client_msg(cur, client_id, store_id, dup, direction, mid, qos, retain, state):
cur.execute('SELECT client_id,store_id,dup,direction,mid,qos,retain,state ' +
'FROM client_msgs')
row = cur.fetchone()
if row[0] != client_id:
raise ValueError("Invalid client_id %s / %s" % (row[0], client_id))
if row[1] != store_id:
raise ValueError("Invalid store_id %d / %d" % (row[1], store_id))
if row[2] != dup:
raise ValueError("Invalid dup %d / %d" % (row[2], dup))
if row[3] != direction:
raise ValueError("Invalid direction %d / %d" % (row[3], direction))
if row[4] != mid:
raise ValueError("Invalid mid %d / %d" % (row[4], mid))
if row[5] != qos:
raise ValueError("Invalid qos %d / %d" % (row[5], qos))
if row[6] != retain:
raise ValueError("Invalid retain %d / %d" % (row[6], retain))
if row[7] != state:
raise ValueError("Invalid state %d / %d" % (row[7], state))
def check_store_msg(cur, expiry_time, topic, payload, source_id, source_username,
payloadlen, source_mid, source_port, qos, retain, idx=0):
cur.execute('SELECT store_id,expiry_time,topic,payload,source_id,source_username, ' +
'payloadlen, source_mid, source_port, qos, retain ' +
'FROM msgs')
for i in range(0, idx+1):
row = cur.fetchone()
if row[0] == 0:
raise ValueError("Invalid store_id %d / %d" % (row[0], store_id))
if (expiry_time == 0 and row[1] != 0) or (expiry_time != 0 and row[1] == 0):
raise ValueError("Invalid expiry_time %d / %d" % (row[1], expiry_time))
if row[2] != topic:
raise ValueError("Invalid topic %s / %s" % (row[2], topic))
if row[3] != payload:
raise ValueError("Invalid payload %s / %s" % (row[3], payload))
if row[4] != source_id:
raise ValueError("Invalid source_id %s / %s" % (row[4], source_id))
if row[5] != source_username:
raise ValueError("Invalid source_username %s / %s" % (row[5], source_username))
if row[6] != payloadlen or (payloadlen != 0 and row[6] != len(row[3])):
raise ValueError("Invalid payloadlen %d / %d" % (row[6], payloadlen))
if row[7] != source_mid:
raise ValueError("Invalid source_mid %d / %d" % (row[7], source_mid))
if row[8] != source_port:
raise ValueError("Invalid source_port %d / %d" % (row[8], source_port))
if row[9] != qos:
raise ValueError("Invalid qos %d / %d" % (row[9], qos))
if row[10] != retain:
raise ValueError("Invalid retain %d / %d" % (row[10], retain))
return row[0]
def check_retain(cur, topic, store_id):
cur.execute('SELECT store_id FROM retains WHERE topic=?', (topic,))
row = cur.fetchone()
if row[0] != store_id:
raise ValueError("Invalid store_id %d / %d" % (row[0], store_id))
+16
View File
@@ -224,6 +224,22 @@ tests = [
#(1, './15-persist-retain-v5-0.py'),
#(1, './15-persist-subscription-v3-1-1.py'),
#(1, './15-persist-subscription-v5-0.py'),
(1, './15-sqlite-clean-shutdown.py'),
(1, './15-sqlite-client-message-in-v3-1-1.py'),
(1, './15-sqlite-client-message-out-dup-v3-1-1.py'),
(1, './15-sqlite-client-message-out-v3-1-1.py'),
(1, './15-sqlite-client-msg-in-v5-0.py'),
(1, './15-sqlite-client-msg-out-queue-v3-1-1.py'),
(1, './15-sqlite-client-msg-out-v3-1-1.py'),
(1, './15-sqlite-client-msg-out-v5-0.py'),
(1, './15-sqlite-client-v3-1-1.py'),
(1, './15-sqlite-client-v5-0.py'),
(1, './15-sqlite-client-v5.0.py'),
(1, './15-sqlite-publish-properties-v5-0.py'),
(1, './15-sqlite-retain-v3-1-1.py'),
(1, './15-sqlite-retain-v5-0.py'),
(1, './15-sqlite-subscription-v3-1-1.py'),
(1, './15-sqlite-subscription-v5-0.py'),
(1, './16-cmd-args.py'),
(1, './16-config-includedir.py'),