From dc906c5221adf4119f6472870824ce4c15cf22de Mon Sep 17 00:00:00 2001 From: "Roger A. Light" Date: Sat, 11 Dec 2021 22:37:41 +0000 Subject: [PATCH] Sqlite persistence plugin. --- config.mk | 3 + plugins/Makefile | 3 +- plugins/persist-sqlite/CMakeLists.txt | 45 +++ plugins/persist-sqlite/Makefile | 68 ++++ plugins/persist-sqlite/client_msgs.c | 133 +++++++ plugins/persist-sqlite/clients.c | 122 ++++++ plugins/persist-sqlite/init.c | 308 +++++++++++++++ plugins/persist-sqlite/msg_store.c | 262 ++++++++++++ plugins/persist-sqlite/persist_sqlite.h | 68 ++++ plugins/persist-sqlite/plugin.c | 161 ++++++++ plugins/persist-sqlite/restore.c | 373 ++++++++++++++++++ plugins/persist-sqlite/retains.c | 71 ++++ plugins/persist-sqlite/subscriptions.c | 86 ++++ plugins/persist-sqlite/test.conf | 2 + plugins/persist-sqlite/test.sh | 1 + src/database.c | 2 +- test/broker/15-sqlite-clean-shutdown.py | 42 ++ .../15-sqlite-client-message-in-v3-1-1.py | 91 +++++ ...15-sqlite-client-message-out-dup-v3-1-1.py | 117 ++++++ .../15-sqlite-client-message-out-v3-1-1.py | 88 +++++ test/broker/15-sqlite-client-msg-in-v5-0.py | 92 +++++ .../15-sqlite-client-msg-out-queue-v3-1-1.py | 102 +++++ .../broker/15-sqlite-client-msg-out-v3-1-1.py | 102 +++++ test/broker/15-sqlite-client-msg-out-v5-0.py | 102 +++++ test/broker/15-sqlite-client-v3-1-1.py | 87 ++++ test/broker/15-sqlite-client-v5-0.py | 69 ++++ test/broker/15-sqlite-client-v5.0.py | 83 ++++ .../15-sqlite-publish-properties-v5-0.py | 86 ++++ test/broker/15-sqlite-retain-v3-1-1.py | 89 +++++ test/broker/15-sqlite-retain-v5-0.py | 88 +++++ test/broker/15-sqlite-subscription-v3-1-1.py | 128 ++++++ test/broker/15-sqlite-subscription-v5-0.py | 139 +++++++ test/broker/Makefile | 15 + test/broker/sqlite_help.py | 217 ++++++++++ test/broker/test.py | 16 + 35 files changed, 3459 insertions(+), 2 deletions(-) create mode 100644 plugins/persist-sqlite/CMakeLists.txt create mode 100644 plugins/persist-sqlite/Makefile create mode 100644 plugins/persist-sqlite/client_msgs.c create mode 100644 plugins/persist-sqlite/clients.c create mode 100644 plugins/persist-sqlite/init.c create mode 100644 plugins/persist-sqlite/msg_store.c create mode 100644 plugins/persist-sqlite/persist_sqlite.h create mode 100644 plugins/persist-sqlite/plugin.c create mode 100644 plugins/persist-sqlite/restore.c create mode 100644 plugins/persist-sqlite/retains.c create mode 100644 plugins/persist-sqlite/subscriptions.c create mode 100644 plugins/persist-sqlite/test.conf create mode 100755 plugins/persist-sqlite/test.sh create mode 100755 test/broker/15-sqlite-clean-shutdown.py create mode 100755 test/broker/15-sqlite-client-message-in-v3-1-1.py create mode 100755 test/broker/15-sqlite-client-message-out-dup-v3-1-1.py create mode 100755 test/broker/15-sqlite-client-message-out-v3-1-1.py create mode 100755 test/broker/15-sqlite-client-msg-in-v5-0.py create mode 100755 test/broker/15-sqlite-client-msg-out-queue-v3-1-1.py create mode 100755 test/broker/15-sqlite-client-msg-out-v3-1-1.py create mode 100755 test/broker/15-sqlite-client-msg-out-v5-0.py create mode 100755 test/broker/15-sqlite-client-v3-1-1.py create mode 100755 test/broker/15-sqlite-client-v5-0.py create mode 100755 test/broker/15-sqlite-client-v5.0.py create mode 100755 test/broker/15-sqlite-publish-properties-v5-0.py create mode 100755 test/broker/15-sqlite-retain-v3-1-1.py create mode 100755 test/broker/15-sqlite-retain-v5-0.py create mode 100755 test/broker/15-sqlite-subscription-v3-1-1.py create mode 100755 test/broker/15-sqlite-subscription-v5-0.py create mode 100755 test/broker/sqlite_help.py diff --git a/config.mk b/config.mk index 04851ac4..3235269b 100644 --- a/config.mk +++ b/config.mk @@ -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 # ============================================================================= diff --git a/plugins/Makefile b/plugins/Makefile index 4e01d1ab..9cf8b229 100644 --- a/plugins/Makefile +++ b/plugins/Makefile @@ -1,6 +1,7 @@ DIRS= \ dynamic-security \ - examples + examples \ + persist-sqlite .PHONY : all binary check clean reallyclean test install uninstall diff --git a/plugins/persist-sqlite/CMakeLists.txt b/plugins/persist-sqlite/CMakeLists.txt new file mode 100644 index 00000000..f58fe755 --- /dev/null +++ b/plugins/persist-sqlite/CMakeLists.txt @@ -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() diff --git a/plugins/persist-sqlite/Makefile b/plugins/persist-sqlite/Makefile new file mode 100644 index 00000000..9dbbccae --- /dev/null +++ b/plugins/persist-sqlite/Makefile @@ -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" diff --git a/plugins/persist-sqlite/client_msgs.c b/plugins/persist-sqlite/client_msgs.c new file mode 100644 index 00000000..b2e5e38c --- /dev/null +++ b/plugins/persist-sqlite/client_msgs.c @@ -0,0 +1,133 @@ +/* +Copyright (c) 2021 Roger Light + +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 +#include + +#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; +} diff --git a/plugins/persist-sqlite/clients.c b/plugins/persist-sqlite/clients.c new file mode 100644 index 00000000..9d91a3cc --- /dev/null +++ b/plugins/persist-sqlite/clients.c @@ -0,0 +1,122 @@ +/* +Copyright (c) 2021 Roger Light + +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 +#include + +#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; +} diff --git a/plugins/persist-sqlite/init.c b/plugins/persist-sqlite/init.c new file mode 100644 index 00000000..d831bb08 --- /dev/null +++ b/plugins/persist-sqlite/init.c @@ -0,0 +1,308 @@ +/* +Copyright (c) 2021 Roger Light + +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 +#include + +#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; + } +} diff --git a/plugins/persist-sqlite/msg_store.c b/plugins/persist-sqlite/msg_store.c new file mode 100644 index 00000000..4a282729 --- /dev/null +++ b/plugins/persist-sqlite/msg_store.c @@ -0,0 +1,262 @@ +/* +Copyright (c) 2021 Roger Light + +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 +#include +#include +#include + +#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; +} diff --git a/plugins/persist-sqlite/persist_sqlite.h b/plugins/persist-sqlite/persist_sqlite.h new file mode 100644 index 00000000..a484d786 --- /dev/null +++ b/plugins/persist-sqlite/persist_sqlite.h @@ -0,0 +1,68 @@ +/* +Copyright (c) 2021 Roger Light + +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 + +#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 diff --git a/plugins/persist-sqlite/plugin.c b/plugins/persist-sqlite/plugin.c new file mode 100644 index 00000000..df3b5ccd --- /dev/null +++ b/plugins/persist-sqlite/plugin.c @@ -0,0 +1,161 @@ +/* +Copyright (c) 2021 Roger Light + +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 +#include +#include +#include +#include + +#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 + +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 +#include +#include +#include + +#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; +} diff --git a/plugins/persist-sqlite/retains.c b/plugins/persist-sqlite/retains.c new file mode 100644 index 00000000..90fdf307 --- /dev/null +++ b/plugins/persist-sqlite/retains.c @@ -0,0 +1,71 @@ +/* +Copyright (c) 2021 Roger Light + +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 +#include + +#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; +} diff --git a/plugins/persist-sqlite/subscriptions.c b/plugins/persist-sqlite/subscriptions.c new file mode 100644 index 00000000..8422461f --- /dev/null +++ b/plugins/persist-sqlite/subscriptions.c @@ -0,0 +1,86 @@ +/* +Copyright (c) 2021 Roger Light + +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 +#include + +#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; +} diff --git a/plugins/persist-sqlite/test.conf b/plugins/persist-sqlite/test.conf new file mode 100644 index 00000000..5253e040 --- /dev/null +++ b/plugins/persist-sqlite/test.conf @@ -0,0 +1,2 @@ +plugin ./mosquitto_persist_sqlite.so +plugin_opt_db_file test.sqlite3 diff --git a/plugins/persist-sqlite/test.sh b/plugins/persist-sqlite/test.sh new file mode 100755 index 00000000..51b979db --- /dev/null +++ b/plugins/persist-sqlite/test.sh @@ -0,0 +1 @@ +../../src/mosquitto -c test.conf -v diff --git a/src/database.c b/src/database.c index fe5154df..8446444f 100644 --- a/src/database.c +++ b/src/database.c @@ -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); } diff --git a/test/broker/15-sqlite-clean-shutdown.py b/test/broker/15-sqlite-clean-shutdown.py new file mode 100755 index 00000000..69a119a8 --- /dev/null +++ b/test/broker/15-sqlite-clean-shutdown.py @@ -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) diff --git a/test/broker/15-sqlite-client-message-in-v3-1-1.py b/test/broker/15-sqlite-client-message-in-v3-1-1.py new file mode 100755 index 00000000..898eb397 --- /dev/null +++ b/test/broker/15-sqlite-client-message-in-v3-1-1.py @@ -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) diff --git a/test/broker/15-sqlite-client-message-out-dup-v3-1-1.py b/test/broker/15-sqlite-client-message-out-dup-v3-1-1.py new file mode 100755 index 00000000..43199230 --- /dev/null +++ b/test/broker/15-sqlite-client-message-out-dup-v3-1-1.py @@ -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) diff --git a/test/broker/15-sqlite-client-message-out-v3-1-1.py b/test/broker/15-sqlite-client-message-out-v3-1-1.py new file mode 100755 index 00000000..e9c5a0d6 --- /dev/null +++ b/test/broker/15-sqlite-client-message-out-v3-1-1.py @@ -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) diff --git a/test/broker/15-sqlite-client-msg-in-v5-0.py b/test/broker/15-sqlite-client-msg-in-v5-0.py new file mode 100755 index 00000000..6d6896c4 --- /dev/null +++ b/test/broker/15-sqlite-client-msg-in-v5-0.py @@ -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) diff --git a/test/broker/15-sqlite-client-msg-out-queue-v3-1-1.py b/test/broker/15-sqlite-client-msg-out-queue-v3-1-1.py new file mode 100755 index 00000000..0c7d354d --- /dev/null +++ b/test/broker/15-sqlite-client-msg-out-queue-v3-1-1.py @@ -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) diff --git a/test/broker/15-sqlite-client-msg-out-v3-1-1.py b/test/broker/15-sqlite-client-msg-out-v3-1-1.py new file mode 100755 index 00000000..0c7d354d --- /dev/null +++ b/test/broker/15-sqlite-client-msg-out-v3-1-1.py @@ -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) diff --git a/test/broker/15-sqlite-client-msg-out-v5-0.py b/test/broker/15-sqlite-client-msg-out-v5-0.py new file mode 100755 index 00000000..7b9a129d --- /dev/null +++ b/test/broker/15-sqlite-client-msg-out-v5-0.py @@ -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) diff --git a/test/broker/15-sqlite-client-v3-1-1.py b/test/broker/15-sqlite-client-v3-1-1.py new file mode 100755 index 00000000..f55ee29c --- /dev/null +++ b/test/broker/15-sqlite-client-v3-1-1.py @@ -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) diff --git a/test/broker/15-sqlite-client-v5-0.py b/test/broker/15-sqlite-client-v5-0.py new file mode 100755 index 00000000..f71410dc --- /dev/null +++ b/test/broker/15-sqlite-client-v5-0.py @@ -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) diff --git a/test/broker/15-sqlite-client-v5.0.py b/test/broker/15-sqlite-client-v5.0.py new file mode 100755 index 00000000..3cdaad8e --- /dev/null +++ b/test/broker/15-sqlite-client-v5.0.py @@ -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) diff --git a/test/broker/15-sqlite-publish-properties-v5-0.py b/test/broker/15-sqlite-publish-properties-v5-0.py new file mode 100755 index 00000000..2314f7f2 --- /dev/null +++ b/test/broker/15-sqlite-publish-properties-v5-0.py @@ -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) diff --git a/test/broker/15-sqlite-retain-v3-1-1.py b/test/broker/15-sqlite-retain-v3-1-1.py new file mode 100755 index 00000000..dcd311cf --- /dev/null +++ b/test/broker/15-sqlite-retain-v3-1-1.py @@ -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) diff --git a/test/broker/15-sqlite-retain-v5-0.py b/test/broker/15-sqlite-retain-v5-0.py new file mode 100755 index 00000000..a5c6e609 --- /dev/null +++ b/test/broker/15-sqlite-retain-v5-0.py @@ -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) diff --git a/test/broker/15-sqlite-subscription-v3-1-1.py b/test/broker/15-sqlite-subscription-v3-1-1.py new file mode 100755 index 00000000..99e0c731 --- /dev/null +++ b/test/broker/15-sqlite-subscription-v3-1-1.py @@ -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) diff --git a/test/broker/15-sqlite-subscription-v5-0.py b/test/broker/15-sqlite-subscription-v5-0.py new file mode 100755 index 00000000..fb7250a0 --- /dev/null +++ b/test/broker/15-sqlite-subscription-v5-0.py @@ -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) diff --git a/test/broker/Makefile b/test/broker/Makefile index 1005ef4c..7c265e31 100644 --- a/test/broker/Makefile +++ b/test/broker/Makefile @@ -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 diff --git a/test/broker/sqlite_help.py b/test/broker/sqlite_help.py new file mode 100755 index 00000000..91dd24b6 --- /dev/null +++ b/test/broker/sqlite_help.py @@ -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)) diff --git a/test/broker/test.py b/test/broker/test.py index 7ba45f4c..b0e46589 100755 --- a/test/broker/test.py +++ b/test/broker/test.py @@ -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'),