Mirrors_Framework/mosquitto
1
0
Fork 0
mirror of https://github.com/eclipse-mosquitto/mosquitto.git synced 2026-09-23 00:24:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'NorbertHeusser-fix-persist-plugin-expiry' into develop

Browse Source
This commit is contained in:
Roger A. Light
2025-02-26 09:59:49 +00:00
parent 68c81c7fc3 f8d37b70a6
commit d45f033be4
21 changed files with 1102 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/persist-sqlite/base_msgs.c
+18
View File
@@ -117,3 +117,21 @@ int persist_sqlite__base_msg_remove_cb(int event, void *event_data, void *userda
return rc;
}
int persist_sqlite__base_msg_clear(struct mosquitto_sqlite *ms, const char *clientid)
{
int rc = MOSQ_ERR_UNKNOWN;
if(sqlite3_bind_text(ms->base_msg_remove_for_clientid_stmt, 1, clientid, (int)strlen(clientid), SQLITE_STATIC) == SQLITE_OK){
ms->event_count++;
rc = sqlite3_step(ms->base_msg_remove_for_clientid_stmt);
if(rc == SQLITE_DONE){
rc = MOSQ_ERR_SUCCESS;
}else{
rc = MOSQ_ERR_UNKNOWN;
}
}
sqlite3_reset(ms->base_msg_remove_for_clientid_stmt);
return rc;
}
plugins/persist-sqlite/clients.c
+6 -1
View File
@@ -89,6 +89,11 @@ int persist_sqlite__client_remove_cb(int event, void *event_data, void *userdata
rc = MOSQ_ERR_UNKNOWN;
}
}
/* Delete base msgs before deletion of client_msgs as the query will iterate over the client_msgs table */
persist_sqlite__base_msg_clear(ms, ed->data.clientid);
persist_sqlite__client_msg_clear(ms, ed->data.clientid);
if(sqlite3_bind_text(ms->client_remove_stmt, 1,
ed->data.clientid, (int)strlen(ed->data.clientid), SQLITE_STATIC) == SQLITE_OK){
@@ -101,7 +106,6 @@ int persist_sqlite__client_remove_cb(int event, void *event_data, void *userdata
rc = MOSQ_ERR_UNKNOWN;
}
}
persist_sqlite__client_msg_clear(ms, ed->data.clientid);
return rc;
}
@@ -133,3 +137,4 @@ int persist_sqlite__client_update_cb(int event, void *event_data, void *userdata
return rc;
}
plugins/persist-sqlite/init.c
+23 -1
View File
@@ -141,8 +141,19 @@ static int create_tables(struct mosquitto_sqlite *ms)
"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);",
"DROP INDEX IF EXISTS client_msgs_store_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,client_id);",
NULL, NULL, NULL);
if(rc) goto fail;
rc = sqlite3_exec(ms->db,
"CREATE INDEX IF NOT EXISTS retains_storeid ON retains(store_id);",
NULL, NULL, NULL);
if(rc) goto fail;
@@ -294,6 +305,17 @@ static int prepare_statements(struct mosquitto_sqlite *ms)
&ms->base_msg_remove_stmt, NULL);
if(rc) goto fail;
rc = sqlite3_prepare_v3(ms->db,
"DELETE FROM base_msgs AS bm "
"WHERE bm.store_id IN "
"( SELECT cm.store_id FROM client_msgs AS cm"
" LEFT OUTER JOIN client_msgs AS oc ON oc.store_id = cm.store_id AND oc.client_id != cm.client_id"
" LEFT OUTER JOIN retains AS rm ON rm.store_id = cm.store_id"
" WHERE cm.client_id = ? AND oc.store_id IS NULL AND rm.store_id IS NULL)",
-1, SQLITE_PREPARE_PERSISTENT,
&ms->base_msg_remove_for_clientid_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 "
plugins/persist-sqlite/persist_sqlite.h
+2
View File
@@ -43,6 +43,7 @@ struct mosquitto_sqlite {
sqlite3_stmt *client_msg_clear_all_stmt;
sqlite3_stmt *base_msg_add_stmt;
sqlite3_stmt *base_msg_remove_stmt;
sqlite3_stmt *base_msg_remove_for_clientid_stmt;
sqlite3_stmt *base_msg_load_stmt;
sqlite3_stmt *retain_msg_set_stmt;
sqlite3_stmt *retain_msg_remove_stmt;
@@ -70,6 +71,7 @@ int persist_sqlite__client_msg_update_cb(int event, void *event_data, void *user
int persist_sqlite__base_msg_add_cb(int event, void *event_data, void *userdata);
int persist_sqlite__base_msg_load_cb(int event, void *event_data, void *userdata);
int persist_sqlite__base_msg_remove_cb(int event, void *event_data, void *userdata);
int persist_sqlite__base_msg_clear(struct mosquitto_sqlite *ms, const char *clientid);
int persist_sqlite__retain_msg_set_cb(int event, void *event_data, void *userdata);
int persist_sqlite__retain_msg_remove_cb(int event, void *event_data, void *userdata);
int persist_sqlite__subscription_add_cb(int event, void *event_data, void *userdata);
src/database.c
+18 -4
View File
@@ -373,6 +373,10 @@ static void db__fill_inflight_out_from_queue(struct mosquitto *context)
client_msg->data.state = mosq_ms_publish_qos2;
break;
}
if(client_msg->base_msg->data.expiry_time && db.now_real_s > client_msg->base_msg->data.expiry_time){
db__message_remove_queued(context, &context->msgs_out, client_msg);
continue;
}
plugin_persist__handle_client_msg_update(context, client_msg);
db__message_dequeue_first(context, &context->msgs_out);
}
@@ -936,10 +940,12 @@ int db__message_store(const struct mosquitto *source, struct mosquitto__base_msg
base_msg->source_listener = source->listener;
}
base_msg->origin = origin;
if(message_expiry_interval && *message_expiry_interval != MSG_EXPIRY_INFINITE){
base_msg->data.expiry_time = db.now_real_s + (*message_expiry_interval);
}else{
base_msg->data.expiry_time = 0;
if(message_expiry_interval){
if(*message_expiry_interval > 0 && *message_expiry_interval != MSG_EXPIRY_INFINITE){
base_msg->data.expiry_time = db.now_real_s + *message_expiry_interval;
}else{
base_msg->data.expiry_time = 0;
}
}
base_msg->dest_ids = NULL;
@@ -1190,6 +1196,12 @@ int db__message_release_incoming(struct mosquitto *context, uint16_t mid)
}
void db__retain_expiry_check()
{
retain__expiry_check(&db.retains);
}
void db__expire_all_messages(struct mosquitto *context)
{
struct mosquitto__client_msg *client_msg, *tmp;
@@ -1202,6 +1214,7 @@ void db__expire_all_messages(struct mosquitto *context)
db__message_remove_inflight(context, &context->msgs_out, client_msg);
}
}
db__fill_inflight_out_from_queue(context);
DL_FOREACH_SAFE(context->msgs_out.queued, client_msg, tmp){
if(client_msg->base_msg->data.expiry_time && db.now_real_s > client_msg->base_msg->data.expiry_time){
db__message_remove_queued(context, &context->msgs_out, client_msg);
@@ -1284,6 +1297,7 @@ static int db__message_write_inflight_out_single(struct mosquitto *context, stru
util__increment_send_quota(context);
}
db__message_remove_inflight(context, &context->msgs_out, client_msg);
db__fill_inflight_out_from_queue(context);
return MOSQ_ERR_SUCCESS;
}else{
expiry_interval = (uint32_t)(base_msg->data.expiry_time - db.now_real_s);
src/mosquitto.c
+2
View File
@@ -514,6 +514,8 @@ int main(int argc, char *argv[])
}
plugin_persist__handle_restore();
session_expiry__check();
db__retain_expiry_check();
db__msg_store_compact();
/* After loading persisted clients and ACLs, try to associate them,
src/mosquitto_broker_internal.h
+2 -1
View File
@@ -737,6 +737,7 @@ int db__message_write_queued_in(struct mosquitto *context);
void db__msg_add_to_inflight_stats(struct mosquitto_msg_data *msg_data, struct mosquitto__client_msg *msg);
void db__msg_add_to_queued_stats(struct mosquitto_msg_data *msg_data, struct mosquitto__client_msg *msg);
uint64_t db__new_msg_id(void);
void db__retain_expiry_check(void);
void db__expire_all_messages(struct mosquitto *context);
void db__check_acl_of_all_messages(struct mosquitto *context);
@@ -895,7 +896,7 @@ int retain__init(void);
void retain__clean(struct mosquitto__retainhier **retainhier);
int retain__queue(struct mosquitto *context, const struct mosquitto_subscription *sub);
int retain__store(const char *topic, struct mosquitto__base_msg *base_msg, char **split_topics, bool persist);
void retain__expiry_check(struct mosquitto__retainhier **retainhier);
/* ============================================================
* Security related functions
* ============================================================ */
src/plugin_public.c
+3 -18
View File
@@ -526,6 +526,7 @@ BROKER_EXPORT int mosquitto_persist_client_add(struct mosquitto_client *client)
}
context__add_to_by_id(context);
session_expiry__add_from_persistence(context,context->session_expiry_time);
return MOSQ_ERR_SUCCESS;
error:
@@ -759,9 +760,6 @@ BROKER_EXPORT int mosquitto_persist_base_msg_add(struct mosquitto_base_msg *msg_
{
struct mosquitto context;
struct mosquitto__base_msg *base_msg;
uint32_t message_expiry_interval;
uint32_t *p_message_expiry_interval;
time_t message_expiry_interval_tt;
int rc;
memset(&context, 0, sizeof(context));
@@ -775,25 +773,12 @@ BROKER_EXPORT int mosquitto_persist_base_msg_add(struct mosquitto_base_msg *msg_
context.id = (char *)msg_add->source_id;
context.username = (char *)msg_add->source_username;
p_message_expiry_interval = &message_expiry_interval;
if(msg_add->expiry_time == 0){
p_message_expiry_interval = NULL;
}else if(msg_add->expiry_time <= db.now_real_s){
message_expiry_interval = 0;
}else{
message_expiry_interval_tt = msg_add->expiry_time - db.now_real_s;
if(message_expiry_interval_tt > UINT32_MAX){
message_expiry_interval = UINT32_MAX;
}else{
message_expiry_interval = (uint32_t)message_expiry_interval_tt;
}
}
base_msg = mosquitto_calloc(1, sizeof(struct mosquitto__base_msg));
if(base_msg == NULL){
goto error;
}
base_msg->data.store_id = msg_add->store_id;
base_msg->data.expiry_time = msg_add->expiry_time;
base_msg->data.payloadlen = msg_add->payloadlen;
base_msg->data.source_mid = msg_add->source_mid;
base_msg->data.qos = msg_add->qos;
@@ -816,7 +801,7 @@ BROKER_EXPORT int mosquitto_persist_base_msg_add(struct mosquitto_base_msg *msg_
}
base_msg->stored = true;
rc = db__message_store(&context, base_msg, p_message_expiry_interval, mosq_mo_broker);
rc = db__message_store(&context, base_msg, NULL, mosq_mo_broker);
return rc;
error:
src/retain.c
+26 -7
View File
@@ -191,6 +191,19 @@ int retain__store(const char *topic, struct mosquitto__base_msg *base_msg, char
return MOSQ_ERR_SUCCESS;
}
static bool retain__delete_expired_msg(struct mosquitto__retainhier *branch)
{
if(branch->retained && branch->retained->data.expiry_time > 0 && db.now_real_s >= branch->retained->data.expiry_time){
plugin_persist__handle_retain_msg_delete(branch->retained);
db__msg_store_ref_dec(&branch->retained);
branch->retained = NULL;
#ifdef WITH_SYS_TREE
db.retained_count--;
#endif
return true;
}
return false;
}
static int retain__process(struct mosquitto__retainhier *branch, struct mosquitto *context, const struct mosquitto_subscription *sub)
{
@@ -199,13 +212,7 @@ static int retain__process(struct mosquitto__retainhier *branch, struct mosquitt
uint16_t mid;
struct mosquitto__base_msg *retained;
if(branch->retained->data.expiry_time > 0 && db.now_real_s >= branch->retained->data.expiry_time){
plugin_persist__handle_retain_msg_delete(branch->retained);
db__msg_store_ref_dec(&branch->retained);
branch->retained = NULL;
#ifdef WITH_SYS_TREE
db.retained_count--;
#endif
if(retain__delete_expired_msg(branch)){
return MOSQ_ERR_SUCCESS;
}
@@ -343,6 +350,17 @@ int retain__queue(struct mosquitto *context, const struct mosquitto_subscription
return MOSQ_ERR_SUCCESS;
}
void retain__expiry_check(struct mosquitto__retainhier **retainhier)
{
struct mosquitto__retainhier *peer, *retainhier_tmp;
HASH_ITER(hh, *retainhier, peer, retainhier_tmp){
retain__expiry_check(&peer->children);
if (retain__delete_expired_msg(peer)){
retain__clean_empty_hierarchy(peer);
}
}
}
void retain__clean(struct mosquitto__retainhier **retainhier)
{
@@ -359,3 +377,4 @@ void retain__clean(struct mosquitto__retainhier **retainhier)
}
}
src/session_expiry.c
+2 -2
View File
@@ -149,9 +149,9 @@ void session_expiry__check(void)
struct mosquitto *context;
time_t timeout;
if(db.now_real_s <= last_check){
if(last_check != 0 && db.now_real_s <= last_check){
if(expiry_list){
/* Next event is the first item of the list, we must set the timeout even if we aren't
/* Next event is the first item of the list, we must set the timeout even if we aren't
* checking the full list */
timeout = (expiry_list->context->session_expiry_time - db.now_real_s) * 1000;
if(timeout <= 0){
test/broker/15-persist-bridge-queue.py
+78 -47
View File
@@ -24,6 +24,7 @@ conf_file_bridge_target = os.path.basename(__file__).replace(
".py", "_bridge_target.conf"
)
def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: dict):
persist_help.write_config(
conf_file, port, additional_config_entries=bridging_add_config
@@ -46,6 +47,18 @@ def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: d
source_id = "persist-bridge-test-publisher"
proto_ver = 4
def gen_pub_packets(idx: int, mid_offset: int):
payload = f"queued message {idx:3}"
publish_packet = mosq_test.gen_publish(
topic,
mid=mid_offset + idx,
qos=qos,
payload=payload.encode("UTF-8"),
proto_ver=proto_ver,
)
puback_packet = mosq_test.gen_puback(mid=mid_offset + idx, proto_ver=proto_ver)
return publish_packet, puback_packet
connect_packet = mosq_test.gen_connect(
client_id, proto_ver=proto_ver, clean_session=False
)
@@ -67,44 +80,56 @@ def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: d
bridge_target_broker = mosq_test.start_broker(
filename=conf_file_bridge_target, use_conf=True, port=bridge_target_port
)
# Connect and send a ping to make sure bridge target broker is up
sock = mosq_test.do_client_connect(
# Connect to the bridge target broker and make a qos1 subscription
sock_bridge_target = mosq_test.do_client_connect(
connect_packet, connack_packet1, timeout=5, port=bridge_target_port
)
mosq_test.do_ping(sock)
sock.close()
mosq_test.do_send_receive(
sock_bridge_target,
subscribe_packet,
suback_packet,
"suback from bridge target",
)
# Now start the broker with the bridge
broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port)
# Connect and send a ping to make sure bridging broker is up
# Connect and send a single message forwarded to the bridge target
sock = mosq_test.do_client_connect(
connect2_packet, connack2_packet, timeout=5, port=port
)
publish_packet, puback_packet = gen_pub_packets(0, mid_offset=3)
mosq_test.do_send_receive(
sock, publish_packet, puback_packet, "puback for first message"
)
# Wait until we have received the message from the bridge target
publish_packet, puback_packet = gen_pub_packets(0, mid_offset=1)
mosq_test.do_receive_send(
sock_bridge_target, publish_packet, puback_packet, "first published message"
)
# Wait for a ping response to make sure the target broker has processed the PUBACK
mosq_test.do_ping(sock_bridge_target)
sock_bridge_target.close()
# Make sure the bridging broker processes a ping, which means the PUBACK from the bridge target for the
# first message was processed as well
mosq_test.do_ping(sock)
# Stop the bridge target broker
(broker_terminate_rc, stde) = mosq_test.terminate_broker(bridge_target_broker)
(broker_terminate_rc, stde2) = mosq_test.terminate_broker(bridge_target_broker)
bridge_target_broker = None
# Publish messages
for i in range(num_messages):
payload = f"queued message {i:3}"
mid = 10 + i
publish_packet = mosq_test.gen_publish(
topic,
mid=mid,
qos=qos,
payload=payload.encode("UTF-8"),
proto_ver=proto_ver,
)
puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver)
publish_packet, puback_packet = gen_pub_packets(idx=i, mid_offset=10)
mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback")
sock.close()
# Terminate the bridging broker
(broker_terminate_rc, stde) = mosq_test.terminate_broker(broker)
(broker_terminate_rc, stde3) = mosq_test.terminate_broker(broker)
broker = None
persist_help.check_counts(
@@ -136,8 +161,8 @@ def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: d
)
# Check client msg
subscriber_mid = 3 + i
cmsg_id = 1 + i
subscriber_mid = 4 + i
cmsg_id = 2 + i
persist_help.check_client_msg(
port,
"upstream-bridge",
@@ -149,7 +174,6 @@ def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: d
qos,
0,
persist_help.ms_queued,
idx=i
)
# Start the bridge target broker
@@ -157,27 +181,17 @@ def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: d
filename=conf_file_bridge_target, use_conf=True, port=bridge_target_port
)
# Connect to the bridge target broker and make a qos1 subscription
# Reconnect to the bridge target broker
sock = mosq_test.do_client_connect(
connect_packet, connack_packet2, timeout=5, port=bridge_target_port
)
mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
# Restart bridging broker
broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port)
# Check, if all message got forwarded through the bridge
for i in range(num_messages):
payload = f"queued message {i:3}"
mid = 1 + i
publish_packet = mosq_test.gen_publish(
topic,
mid=mid,
qos=qos,
payload=payload.encode("UTF-8"),
proto_ver=proto_ver,
)
puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver)
publish_packet, puback_packet = gen_pub_packets(idx=i, mid_offset=1)
mosq_test.do_receive_send(
sock,
publish_packet,
@@ -189,11 +203,19 @@ def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: d
mosq_test.do_ping(sock)
sock.close()
# Reconnect to the bridging broker and send ping to make sure the broker has process
# PUBACK from bridge target before getting shut down
sock = mosq_test.do_client_connect(
connect2_packet, connack2_packet, timeout=5, port=port
)
mosq_test.do_ping(sock)
sock.close()
# Stop both brokers
(broker_terminate_rc, stde) = mosq_test.terminate_broker(broker)
broker = None
(broker_terminate_rc, stde2) = mosq_test.terminate_broker(bridge_target_broker)
bridge_target_broker = None
(broker_terminate_rc, stde) = mosq_test.terminate_broker(broker)
broker = None
persist_help.check_counts(
port,
@@ -224,23 +246,32 @@ def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: d
print(f"{test_case_name}")
if rc:
print(stde.decode("utf-8"))
if stde2 is not None:
print("Bridge target brocker log:")
print(stde2.decode("utf-8"))
if stde3 is not None:
print("Bridging brocker log (first run):")
print(stde3.decode("utf-8"))
if stde is not None:
print("Bridging brocker log:")
print(stde.decode("utf-8"))
assert rc == 0, f"rc: {rc}"
in_bridge_config = {
"connection": "in-bridge",
"address": f"localhost:{port}",
"local_clientid": "bridge-test",
"remote_clientid": "upstream-bridge",
"topic": f"{topic} in 2",
"connection": "in-bridge",
"address": f"localhost:{port}",
"local_clientid": "bridge-test",
"remote_clientid": "upstream-bridge",
"topic": f"{topic} in 2",
}
out_bridge_config = {
"connection": "out-bridge",
"address": f"localhost:{bridge_target_port}",
"local_clientid": "upstream-bridge",
"remote_clientid": "bridge-test",
"topic": f"{topic} out 2",
"connection": "out-bridge",
"address": f"localhost:{bridge_target_port}",
"local_clientid": "upstream-bridge",
"remote_clientid": "bridge-test",
"topic": f"{topic} out 2",
}
memory_queue_config = {
@@ -250,7 +281,7 @@ memory_queue_config = {
do_test(
"memory queue out bridge",
bridging_add_config= memory_queue_config | out_bridge_config,
bridging_add_config=memory_queue_config | out_bridge_config,
target_add_config={},
)
test/broker/15-persist-client-drop-expired-messages.py
Executable
+195
View File
@@ -0,0 +1,195 @@
#!/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 *
from persist_module_helper import *
persist_help = persist_module()
port = mosq_test.get_port()
num_messages = 100
proto_ver = 5
qos = 1
topic = "test-expired-msgs"
username = "test-message-expiry"
subscriber_id = "test-subscriber"
second_subscriber_id = "second-subscriber"
publisher_id = "test-publisher"
def do_test(
test_case_name: str,
additional_config_entries: dict,
resubscribe: bool,
num_messages_two_subscribers: int = 0,
num_retain_messages : int = 0,
):
print(
f"{test_case_name}, resubscribe = {resubscribe}, two_subscribers = {'True' if num_messages_two_subscribers > 0 else 'False'}, num_retain_messages = {num_retain_messages} "
)
conf_file = os.path.basename(__file__).replace(".py", f"_{port}.conf")
persist_help.write_config(
conf_file,
port,
additional_config_entries=additional_config_entries,
)
persist_help.init(port)
connect2_packet = mosq_test.gen_connect(
publisher_id, username=username, proto_ver=proto_ver
)
rc = 1
broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port)
con = None
try:
msg_counts = {subscriber_id: num_messages}
connect_client(
port,
subscriber_id,
username,
proto_ver,
session_expiry=60,
subscribe_topic=topic,
).close()
publisher_sock = connect_client(
port, publisher_id, username, proto_ver, session_expiry=0
)
publish_messages(
publisher_sock,
proto_ver,
topic,
0,
num_messages - num_messages_two_subscribers,
message_expiry=60,
retain_end=num_retain_messages,
)
if num_messages_two_subscribers > 0:
msg_counts[second_subscriber_id] = num_messages_two_subscribers
connect_client(
port,
second_subscriber_id,
username,
proto_ver,
session_expiry=60,
subscribe_topic=topic,
).close()
publish_messages(
publisher_sock,
proto_ver,
topic,
num_messages - num_messages_two_subscribers,
num_messages,
message_expiry=60,
retain_end=num_retain_messages,
)
publisher_sock.close()
# Terminate the broker
(broker_terminate_rc, stde) = mosq_test.terminate_broker(broker)
broker = None
check_db(
persist_help,
port,
username,
subscription_topic=topic,
client_msg_counts=msg_counts,
publisher_id=publisher_id,
num_published_msgs=num_messages,
retain_end = num_retain_messages,
message_expiry=60,
)
# Put session expiry_time into the past
assert persist_help.modify_base_msgs(port, sub_expiry_time=120) == num_messages
# Restart broker
broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port)
# Reconnect client(s), it should have a session, but all queued messages should be dropped
for client_id in msg_counts.keys():
subscriber_sock = connect_client(
port,
client_id,
username,
proto_ver,
session_expiry=60,
session_present=True,
subscribe_topic=topic if resubscribe else None,
)
# Send ping and wait for the PINGRESP to make sure the broker will not send a queued message instead
mosq_test.do_ping(subscriber_sock)
subscriber_sock.close()
(broker_terminate_rc, stde) = mosq_test.terminate_broker(broker)
broker = None
for client_id in msg_counts.keys():
# None for subscriber with subscriber_id means no subscription
msg_counts[client_id] = 0
check_db(
persist_help,
port,
username,
subscription_topic=topic,
client_msg_counts=msg_counts,
publisher_id=publisher_id,
num_published_msgs=num_messages,
retain_end = 0,
)
rc = broker_terminate_rc
finally:
if broker is not None:
broker.terminate()
if mosq_test.wait_for_subprocess(broker):
if rc == 0:
rc = 1
(_, stde) = broker.communicate()
os.remove(conf_file)
rc += persist_help.cleanup(port)
if rc:
print(stde.decode("utf-8"))
assert rc == 0, f"rc: {rc}"
memory_queue_config = {
"log_type": "all",
"max_queued_messages": num_messages,
}
do_test(
"memory queue",
additional_config_entries=memory_queue_config,
resubscribe=False,
)
do_test(
"memory queue",
additional_config_entries=memory_queue_config,
resubscribe=True,
)
do_test(
"memory queue",
additional_config_entries=memory_queue_config,
resubscribe=False,
num_messages_two_subscribers=30,
)
do_test(
"memory queue",
additional_config_entries=memory_queue_config,
resubscribe=False,
num_retain_messages=40,
)
test/broker/15-persist-client-expired-session.py
Executable
+253
View File
@@ -0,0 +1,253 @@
#!/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 *
from persist_module_helper import *
persist_help = persist_module()
port = mosq_test.get_port()
num_messages = 100
proto_ver = 5
qos = 1
topic = "test-session-expiry"
username = "test-session-expiry"
subscriber_id = "test-expired-session-subscriber"
second_subscriber_id = "second-subscriber"
publisher_id = "test-expired-session-publisher"
def do_test(
test_case_name: str,
additional_config_entries: dict,
resubscribe: bool,
num_messages_two_subscribers: int = 0,
num_retain_messages : int = 0,
):
print(
f"{test_case_name}, resubscribe = {resubscribe}, two_subscribers = {'True' if num_messages_two_subscribers > 0 else 'False'}, num_retain_messages = {num_retain_messages} "
)
conf_file = os.path.basename(__file__).replace(".py", f"_{port}.conf")
persist_help.write_config(
conf_file,
port,
additional_config_entries=additional_config_entries,
)
persist_help.init(port)
connect2_packet = mosq_test.gen_connect(
publisher_id, username=username, proto_ver=proto_ver
)
rc = 1
broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port)
con = None
try:
msg_counts = {subscriber_id: num_messages}
connect_client(
port,
subscriber_id,
username,
proto_ver,
session_expiry=60,
subscribe_topic=topic,
).close()
publisher_sock = connect_client(
port, publisher_id, username, proto_ver, session_expiry=0
)
publish_messages(
publisher_sock,
proto_ver,
topic,
0,
num_messages - num_messages_two_subscribers,
retain_end=num_retain_messages,
)
if num_messages_two_subscribers > 0:
msg_counts[second_subscriber_id] = num_messages_two_subscribers
connect_client(
port,
second_subscriber_id,
username,
proto_ver,
session_expiry=60,
subscribe_topic=topic,
).close()
publish_messages(
publisher_sock,
proto_ver,
topic,
num_messages - num_messages_two_subscribers,
num_messages,
retain_end=num_retain_messages,
)
publisher_sock.close()
# Terminate the broker
(broker_terminate_rc, stde) = mosq_test.terminate_broker(broker)
broker = None
check_db(
persist_help,
port,
username,
subscription_topic=topic,
client_msg_counts=msg_counts,
publisher_id=publisher_id,
num_published_msgs=num_messages,
retain_end = num_retain_messages,
)
# Put session expiry_time into the past
assert persist_help.modify_client(port, subscriber_id, sub_expiry_time=120) == 1
# Restart broker
broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port)
# Reconnect client, it should have a session, but all queued messages should be dropped
subscriber_sock = connect_client(
port,
subscriber_id,
username,
proto_ver,
session_expiry=60,
subscribe_topic=topic if resubscribe else None,
)
# Send ping and wait for the PINGRESP to make sure the broker will not send a queued message instead
mosq_test.do_ping(subscriber_sock)
subscriber_sock.close()
(broker_terminate_rc, stde) = mosq_test.terminate_broker(broker)
broker = None
# None for subscriber with subscriber_id means no subscription
msg_counts[subscriber_id] = 0 if resubscribe else None
check_db(
persist_help,
port,
username,
subscription_topic=topic,
client_msg_counts=msg_counts,
publisher_id=publisher_id,
num_published_msgs=num_messages,
retain_end=num_retain_messages,
)
if num_messages_two_subscribers > 0:
# Put session expiry_time into the past
assert (
persist_help.modify_client(
port, second_subscriber_id, sub_expiry_time=120
)
== 1
)
# Restart broker
broker = mosq_test.start_broker(
filename=conf_file, use_conf=True, port=port
)
# Reconnect client, it should have a session, but all queued messages should be dropped
subscriber_sock = connect_client(
port,
second_subscriber_id,
username,
proto_ver,
session_expiry=60,
subscribe_topic=topic if resubscribe else None,
)
# Send ping and wait for the PINGRESP to make sure the broker will not send a queued message instead
mosq_test.do_ping(subscriber_sock)
subscriber_sock.close()
(broker_terminate_rc, stde) = mosq_test.terminate_broker(broker)
broker = None
msg_counts[second_subscriber_id] = 0 if resubscribe else None
check_db(
persist_help,
port,
username,
subscription_topic=topic,
client_msg_counts=msg_counts,
publisher_id=publisher_id,
num_published_msgs=num_messages,
retain_end=num_retain_messages,
)
rc = broker_terminate_rc
finally:
if broker is not None:
broker.terminate()
if mosq_test.wait_for_subprocess(broker):
if rc == 0:
rc = 1
(_, stde) = broker.communicate()
os.remove(conf_file)
rc += persist_help.cleanup(port)
if rc:
print(stde.decode("utf-8"))
assert rc == 0, f"rc: {rc}"
memory_queue_config = {
"log_type": "all",
"max_queued_messages": num_messages,
}
do_test(
"memory queue",
additional_config_entries=memory_queue_config,
resubscribe=False,
)
do_test(
"memory queue",
additional_config_entries=memory_queue_config,
resubscribe=True,
)
do_test(
"memory queue",
additional_config_entries=memory_queue_config,
resubscribe=False,
num_messages_two_subscribers=20,
)
do_test(
"memory queue",
additional_config_entries=memory_queue_config,
resubscribe=True,
num_messages_two_subscribers=20,
)
do_test(
"memory queue",
additional_config_entries=memory_queue_config,
resubscribe=False,
num_retain_messages=30,
)
# The following test case is open for discussion as adapting
# the check routines will be hard and some observations
# are unclear right now
# do_test(
# "memory queue",
# additional_config_entries=memory_queue_config,
# resubscribe=False,
# num_messages_two_subscribers=40,
# num_retain_messages=30,
# )
# do_test(
# "memory queue",
# additional_config_entries=memory_queue_config,
# resubscribe=False,
# num_messages_two_subscribers=50,
# num_retain_messages=60,
# )
test/broker/15-persist-client-msg-modify-acl.py
+2 -7
View File
@@ -53,9 +53,7 @@ def do_test(test_case_name: str, additional_config_entries: dict):
rc = 1
broker = mosq_test.start_broker(
filename=os.path.basename(__file__), use_conf=True, port=port
)
broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port)
con = None
try:
@@ -146,7 +144,6 @@ def do_test(test_case_name: str, additional_config_entries: dict):
qos,
0,
persist_help.ms_queued,
idx=i,
)
# Remove any permission for the test topic and the test user
@@ -156,9 +153,7 @@ def do_test(test_case_name: str, additional_config_entries: dict):
os.chmod(f"{acl_file}", 0o644)
# Restart broker
broker = mosq_test.start_broker(
filename=os.path.basename(__file__), use_conf=True, port=port
)
broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port)
# Connect client again, it should have a session, but all queued messages should be dropped
sock = mosq_test.do_client_connect(
test/broker/Makefile
+2
View File
@@ -276,6 +276,8 @@ endif
PERSIST_TESTS = \
./15-persist-bridge-queue.py \
./15-persist-client-drop-expired-messages.py \
./15-persist-client-expired-session.py \
./15-persist-client-msg-in-v3-1-1.py \
./15-persist-client-msg-in-v5-0.py \
./15-persist-client-msg-modify-acl.py \
test/broker/mosq_test_helper.py
+39 -15
View File
@@ -1,7 +1,11 @@
import inspect, os, sys
# From http://stackoverflow.com/questions/279237/python-import-a-module-from-a-folder
cmd_subfolder = os.path.realpath(os.path.abspath(os.path.join(os.path.split(inspect.getfile( inspect.currentframe() ))[0],"..")))
cmd_subfolder = os.path.realpath(
os.path.abspath(
os.path.join(os.path.split(inspect.getfile(inspect.currentframe()))[0], "..")
)
)
if cmd_subfolder not in sys.path:
sys.path.insert(0, cmd_subfolder)
@@ -23,48 +27,67 @@ ssl_dir = source_dir.parent / "ssl"
import importlib
def persist_module():
if len(sys.argv) > 1:
mod = sys.argv.pop(1)
mod = sys.argv.pop(1).replace(".py", "")
else:
raise RuntimeError("Not enough command line arguments - need persist module")
return importlib.import_module(mod)
def do_test_broker_failure(conf_file : str, config : list, port : int, rc_expected : int, error_log_entry : str = None, stdout_entry : str =None, cmd_args : list = None):
def do_test_broker_failure(
conf_file: str,
config: list,
port: int,
rc_expected: int,
error_log_entry: str = None,
stdout_entry: str = None,
cmd_args: list = None,
):
rc = 1
use_conf_file = len(conf_file)
create_conf_file = use_conf_file and len(config)
if create_conf_file:
with open(conf_file, 'w') as f:
with open(conf_file, "w") as f:
f.write("\n".join(config))
f.write("\n")
try:
broker = None
broker = mosq_test.start_broker(conf_file, port=port, use_conf=use_conf_file, expect_fail=True, cmd_args=cmd_args)
broker = mosq_test.start_broker(
conf_file,
port=port,
use_conf=use_conf_file,
expect_fail=True,
cmd_args=cmd_args,
)
(stdo, stde) = broker.communicate()
if broker.returncode != rc_expected:
print(f"Expected broker return code {rc_expected}, got {broker.returncode}")
(stdo, stde) = broker.communicate()
print(stde.decode('utf-8'))
print(stde.decode("utf-8"))
return rc
if error_log_entry is not None:
error_log = stde.decode('utf-8')
error_log = stde.decode("utf-8")
if error_log_entry not in error_log:
print(f"Error log entry: '{error_log_entry}' not found in '{error_log}'")
print(
f"Error log entry: '{error_log_entry}' not found in '{error_log}'"
)
return rc
if stdout_entry is not None:
stdout_log = stdo.decode('utf-8')
stdout_log = stdo.decode("utf-8")
if stdout_entry not in stdout_log:
print(f"Error stdout entry: '{stdout_entry}' not found in '{stdout_log}'")
print(
f"Error stdout entry: '{stdout_entry}' not found in '{stdout_log}'"
)
return rc
rc = 0
except subprocess.TimeoutExpired:
if broker is not None:
mosq_test.wait_for_subprocess(broker,timeout=1)
mosq_test.wait_for_subprocess(broker, timeout=1)
return rc
except Exception as e:
print(e)
@@ -76,8 +99,9 @@ def do_test_broker_failure(conf_file : str, config : list, port : int, rc_expect
except FileNotFoundError:
pass
if rc:
print(f"While testing 'config {chr(10).join(config) if len(config) else ''}'{', args'+ ' '.join(cmd_args) if cmd_args is not None else ''}")
print(
f"While testing 'config {chr(10).join(config) if len(config) else ''}'{', args'+ ' '.join(cmd_args) if cmd_args is not None else ''}"
)
exit(rc)
return rc;
return rc
test/broker/persist_module_helper.py
+161
View File
@@ -0,0 +1,161 @@
#!/usr/bin/env python3
import socket
import mosq_test
import mqtt5_props
from typing import Optional
from types import ModuleType
def connect_client(
port: int,
client_id: str,
username: str,
proto_ver: int,
session_expiry: int,
session_present: bool = False,
subscribe_topic: Optional[str] = None,
qos: int = 1,
):
connect_packet = mosq_test.gen_connect(
client_id=client_id,
username=username,
proto_ver=proto_ver,
clean_session=False,
session_expiry=session_expiry,
)
connack_packet = mosq_test.gen_connack(
rc=0, proto_ver=proto_ver, flags=1 if session_present else 0
)
sock = mosq_test.do_client_connect(
connect_packet, connack_packet, timeout=5, port=port
)
if subscribe_topic is not None:
mid = 1
subscribe_packet = mosq_test.gen_subscribe(
mid, subscribe_topic, qos, proto_ver=proto_ver
)
suback_packet = mosq_test.gen_suback(mid, qos=qos, proto_ver=proto_ver)
mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
return sock
def publish_messages(
sock: socket,
proto_ver: int,
topic: str,
start: int,
end: int,
retain_end=0,
message_expiry: int = 0,
qos: int = 1,
):
for i in range(start, end):
payload = f"queued message {i:3}"
mid = 10 + i
props = (
mqtt5_props.gen_uint32_prop(
mqtt5_props.PROP_MESSAGE_EXPIRY_INTERVAL, message_expiry
)
if message_expiry > 0
else b""
)
publish_packet = mosq_test.gen_publish(
topic,
mid=mid,
qos=qos,
payload=payload.encode("UTF-8"),
retain = True if i < retain_end else False,
proto_ver=proto_ver,
properties=props,
)
puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver)
mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback")
def check_db(
persist_help: ModuleType,
port: int,
username: str,
subscription_topic: str,
client_msg_counts: dict[str, int],
publisher_id: str,
num_published_msgs: int,
retain_end: int,
message_expiry: int = 0,
qos: int = 1,
):
count_list = [v for v in client_msg_counts.values() if v is not None] + [0]
num_base_msgs = max(count_list)
num_subscriptions = sum(1 for c in client_msg_counts.values() if c is not None)
num_client_msgs_out = sum(count_list)
persist_help.check_counts(
port,
clients=len(client_msg_counts),
client_msgs_out=num_client_msgs_out,
base_msgs=num_base_msgs if num_base_msgs > 0 or retain_end == 0 else 1,
retain_msgs=1 if retain_end > 0 else 0,
subscriptions=num_subscriptions,
)
# Check client
for client_id, num_messages_for_client in client_msg_counts.items():
persist_help.check_client(
port,
client_id,
username=username,
will_delay_time=0,
session_expiry_time=60,
listener_port=None, # persist-lmdb reset listener port to 0 on disconnect
max_packet_size=0,
max_qos=2,
retain_available=1,
session_expiry_interval=60,
will_delay_interval=0,
)
# Check subscription
if num_messages_for_client is not None:
persist_help.check_subscription(port, client_id, subscription_topic, qos, 0)
# Check stored message
for i in range(num_base_msgs):
msg_id = num_published_msgs - num_base_msgs + i
payload = f"queued message {msg_id:3}"
payload_b = payload.encode("UTF-8")
mid = 10 + msg_id
store_id = persist_help.check_base_msg(
port,
message_expiry,
subscription_topic,
payload_b,
publisher_id,
username,
len(payload_b),
mid,
port,
qos,
retain=1 if i < retain_end else 0,
idx=i,
)
# Check client msg
for client_id, num_messages_for_client in client_msg_counts.items():
if num_messages_for_client is None:
continue
client_msg_start = num_published_msgs - num_messages_for_client
if msg_id < client_msg_start:
continue
cmsg_id = 1 + msg_id - client_msg_start
subscriber_mid = cmsg_id
persist_help.check_client_msg(
port,
client_id,
cmsg_id,
store_id,
0,
persist_help.dir_out,
subscriber_mid,
qos,
0,
persist_help.ms_queued,
)
test/broker/persist_sqlite.py
+96 -22
View File
@@ -20,7 +20,8 @@ ms_send_pubrec = 10
ms_queued = 11
def write_config(filename, port, additional_config_entries : dict = {}):
def write_config(filename, port, additional_config_entries: dict = {}):
with open(filename, "w") as f:
f.write("listener %d\n" % (port))
f.write("allow_anonymous true\n")
@@ -28,7 +29,7 @@ def write_config(filename, port, additional_config_entries : dict = {}):
f"plugin {mosq_test.get_build_root()}/plugins/persist-sqlite/mosquitto_persist_sqlite.so\n"
)
f.write("plugin_opt_db_file %d/mosquitto.sqlite3\n" % (port))
for entry, value in additional_config_entries.items():
for entry, value in additional_config_entries.items():
f.write(f"{entry} {value}\n")
@@ -78,6 +79,7 @@ def init(port, create_db_of_version: list[int] = None):
# We need to set write permission to everybody as broker will start with privilege drop
os.chmod(f"{port}/mosquitto.sqlite3", 0o666)
def cleanup(port):
rc = 1
try:
@@ -118,6 +120,7 @@ def check_version_infos(port, database_schema_version):
assert row[i] == database_schema_version[i]
con.close()
def check_counts(
port,
clients=0,
@@ -190,10 +193,14 @@ def check_client(
"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"
+ "FROM clients "
+ f"WHERE client_id = '{client_id}'"
)
row = cur.fetchone()
if row is None:
raise ValueError(f"Cannot find client {client_id} in db")
if row[0] != client_id:
raise ValueError("Invalid client_id %s / %s" % (row[0], client_id))
@@ -237,6 +244,25 @@ def check_client(
con.close()
def modify_client(port: int, client_id: str, sub_expiry_time: int):
num_modified_rows = 0
con = sqlite3.connect(f"{port}/mosquitto.sqlite3")
try:
cur = con.cursor()
cur.execute(
"UPDATE clients"
+ f" SET session_expiry_time = session_expiry_time - {sub_expiry_time}"
+ f" WHERE client_id = ?",
(client_id,),
)
num_modified_rows = cur.rowcount
con.commit()
finally:
con.close()
return num_modified_rows
def check_subscription(
port, client_id, topic, subscription_options, subscription_identifier
):
@@ -244,10 +270,14 @@ def check_subscription(
cur = con.cursor()
cur.execute(
"SELECT client_id, topic, subscription_options, subscription_identifier "
+ "FROM subscriptions"
+ "FROM subscriptions "
+ f"WHERE client_id = '{client_id}'"
)
row = cur.fetchone()
if row is None:
raise ValueError(f"Cannot find client {client_id} in db")
if row[0] != client_id:
raise ValueError("Invalid client_id %s / %s" % (row[0], client_id))
@@ -268,7 +298,7 @@ def check_subscription(
def check_client_msg(
port, client_id, cmsg_id, store_id, dup, direction, mid, qos, retain, state, idx=0
port, client_id, cmsg_id, store_id, dup, direction, mid, qos, retain, state
):
con = sqlite3.connect(f"{port}/mosquitto.sqlite3")
try:
@@ -276,39 +306,60 @@ def check_client_msg(
cur.execute(
"SELECT client_id,cmsg_id,store_id,dup,direction,mid,qos,retain,state "
+ "FROM client_msgs "
+ "ORDER BY cmsg_id"
+ f"WHERE client_id = '{client_id}' AND cmsg_id = {cmsg_id}"
)
for i in range(0, idx + 1):
row = cur.fetchone()
row = cur.fetchone()
msg_id = f"client_id={client_id},cmsg_id={cmsg_id}"
if row is None:
raise ValueError(
f"Cannot find client message client_id = {client_id} cmsg_id = {msg_id} in db."
)
if row[0] != client_id:
raise ValueError("Invalid client_id %s / %s" % (row[0], client_id))
raise ValueError(
"Invalid client_id %s / %s for message %s" % (row[0], client_id, msg_id)
)
if row[1] != cmsg_id:
raise ValueError("Invalid cmsg_id %s / %s" % (row[1], cmsg_id))
raise ValueError(
"Invalid cmsg_id %s / %s for message %s" % (row[1], cmsg_id, msg_id)
)
if row[2] != store_id:
raise ValueError("Invalid store_id %d / %d" % (row[2], store_id))
raise ValueError(
"Invalid store_id %d / %d for message %s" % (row[2], store_id, msg_id)
)
if row[3] != dup:
raise ValueError("Invalid dup %d / %d" % (row[3], dup))
raise ValueError(
"Invalid dup %d / %d for message %s" % (row[3], dup, msg_id)
)
if row[4] != direction:
raise ValueError("Invalid direction %d / %d" % (row[4], direction))
raise ValueError(
"Invalid direction %d / %d for message %s" % (row[4], direction, msg_id)
)
if row[5] != mid:
raise ValueError("Invalid mid %d / %d" % (row[5], mid))
raise ValueError(
"Invalid mid %d / %d for message %s" % (row[5], mid, msg_id)
)
if row[6] != qos:
raise ValueError("Invalid qos %d / %d" % (row[6], qos))
raise ValueError(
"Invalid qos %d / %d for message %s" % (row[6], qos, msg_id)
)
if row[7] != retain:
raise ValueError("Invalid retain %d / %d" % (row[7], retain))
raise ValueError(
"Invalid retain %d / %d for message %s" % (row[7], retain, msg_id)
)
if row[8] != state:
raise ValueError("Invalid state %d / %d" % (row[8], state))
except ValueError as err:
raise ValueError(str(err)+ f" at index {idx}") from err
raise ValueError(
"Invalid state %d / %d for message %s" % (row[8], state, msg_id)
)
finally:
con.close()
@@ -337,7 +388,10 @@ def check_base_msg(
)
for i in range(0, idx + 1):
row = cur.fetchone()
if row is None:
raise ValueError(f"no base messages")
if row[0] == 0:
raise ValueError("Invalid store_id %d / %d" % (row[0], store_id))
@@ -354,7 +408,9 @@ def check_base_msg(
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))
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))
@@ -371,13 +427,31 @@ def check_base_msg(
if row[10] != retain:
raise ValueError("Invalid retain %d / %d" % (row[10], retain))
except ValueError as err:
raise ValueError(str(err)+ f" at index {idx}") from err
raise ValueError(str(err) + f" at index {idx}") from err
finally:
con.close()
return row[0]
def modify_base_msgs(
port: int,
sub_expiry_time: int,
):
num_modified_rows = 0
con = sqlite3.connect(f"{port}/mosquitto.sqlite3")
try:
cur = con.cursor()
cur.execute(
"UPDATE base_msgs" + f" SET expiry_time = expiry_time - {sub_expiry_time}"
)
num_modified_rows = cur.rowcount
con.commit()
finally:
con.close()
return num_modified_rows
def check_retain(port, topic, store_id):
con = sqlite3.connect(f"{port}/mosquitto.sqlite3")
cur = con.cursor()
test/mosq_test.py
+6 -6
View File
@@ -417,12 +417,12 @@ def to_string(packet):
return s
elif cmd == 0x20:
# CONNACK
if len(packet) == 4:
(cmd, rl, resv, rc) = struct.unpack('!BBBB', packet)
return "CONNACK, rl="+str(rl)+", res="+str(resv)+", rc="+str(rc)
elif len(packet) == 5:
(cmd, rl, flags, reason_code, proplen) = struct.unpack('!BBBBB', packet)
return "CONNACK, rl="+str(rl)+", flags="+str(flags)+", rc="+str(reason_code)+", proplen="+str(proplen)
if len(packet) >= 4:
(cmd, rl, flags, reason_code) = struct.unpack('!BBBB', packet[0:4])
s=f"CONNACK, rl={rl}, res/flags={flags}, rc={reason_code}"
if len(packet) > 4:
s = s+ f", properties={mqtt5_props.print_properties(packet[4:])}"
return s
else:
return "CONNACK, (not decoded)"
test/mqtt5_props.py
+163
View File
@@ -68,6 +68,169 @@ def pack_varint(varint):
if varint == 0:
return s
def prop_finalise(props):
return pack_varint(len(props)) + props
def gen_properties(properties_dict: list[dict]):
props = b""
if properties_dict is None:
return props
for prop in properties_dict:
id = prop.get("identifier")
value = prop.get("value")
if id in (
PROP_PAYLOAD_FORMAT_INDICATOR,
PROP_REQUEST_PROBLEM_INFO,
PROP_REQUEST_RESPONSE_INFO,
PROP_MAXIMUM_QOS,
PROP_RETAIN_AVAILABLE,
PROP_WILDCARD_SUB_AVAILABLE,
PROP_SUBSCRIPTION_ID_AVAILABLE,
PROP_SHARED_SUB_AVAILABLE,
):
props += gen_byte_prop(id, value)
elif id in (
PROP_MESSAGE_EXPIRY_INTERVAL,
PROP_SESSION_EXPIRY_INTERVAL,
PROP_WILL_DELAY_INTERVAL,
PROP_MAXIMUM_PACKET_SIZE,
):
props += gen_uint32_prop(id, value)
elif id in (
PROP_CONTENT_TYPE,
PROP_RESPONSE_TOPIC,
PROP_CORRELATION_DATA,
PROP_ASSIGNED_CLIENT_IDENTIFIER,
PROP_AUTHENTICATION_METHOD,
PROP_AUTHENTICATION_DATA,
PROP_RESPONSE_INFO,
PROP_SERVER_REFERENCE,
PROP_REASON_STRING,
):
props += gen_string_prop(id, value)
elif id == PROP_SUBSCRIPTION_IDENTIFIER:
props += gen_varint_prop(id, value)
elif id in (
PROP_SERVER_KEEP_ALIVE,
PROP_RECEIVE_MAXIMUM,
PROP_TOPIC_ALIAS_MAXIMUM,
PROP_TOPIC_ALIAS,
):
props += gen_uint16_prop(id, value)
elif id == PROP_USER_PROPERTY:
props += gen_string_pair_prop(id, prop["name"], value)
return props
def unpack_varint(b: bytes):
def decode_len(b: bytes):
for i in range(len(b)):
if b[i] & 0x80 == 0:
return i + 1
return 0
var_len = decode_len(b)
variant = 0
for i in range(var_len - 1, -1, -1):
variant = 0x80 * variant + (struct.unpack("!B", b[i : i + 1])[0] & 0x7F)
return variant, var_len
def unpack_string(b: bytes):
str_len = struct.unpack("!B", b[0:1])[0]
str_value = struct.unpack(f"!{str_len}s", b[1 : str_len + 1])[0]
return str_value, str_len + 1
def unpack_property(b: bytes):
id = struct.unpack("!B", b[0:1])[0]
if id == PROP_PAYLOAD_FORMAT_INDICATOR:
return "PAYLOAD_FORMAT_INDICATOR", struct.unpack("!B", b[1:2])[0], 2
elif id == PROP_PAYLOAD_FORMAT_INDICATOR:
return "PAYLOAD_FORMAT_INDICATOR", struct.unpack("!B", b[1:2])[0], 2
elif id == PROP_REQUEST_PROBLEM_INFO:
return "REQUEST_PROBLEM_INFO", struct.unpack("!B", b[1:2])[0], 2
elif id == PROP_REQUEST_RESPONSE_INFO:
return "REQUEST_RESPONSE_INFO", struct.unpack("!B", b[1:2])[0], 2
elif id == PROP_MAXIMUM_QOS:
return "MAXIMUM_QOS", struct.unpack("!B", b[1:2])[0], 2
elif id == PROP_RETAIN_AVAILABLE:
return "RETAIN_AVAILABLE", struct.unpack("!B", b[1:2])[0], 2
elif id == PROP_WILDCARD_SUB_AVAILABLE:
return "WILDCARD_SUB_AVAILABLE", struct.unpack("!B", b[1:2])[0], 2
elif id == PROP_SUBSCRIPTION_ID_AVAILABLE:
return "SUBSCRIPTION_ID_AVAILABLE", struct.unpack("!B", b[1:2])[0], 2
elif id == PROP_SHARED_SUB_AVAILABLE:
return "SHARED_SUB_AVAILABLE", struct.unpack("!B", b[1:2])[0], 2
elif id == PROP_SERVER_KEEP_ALIVE:
return "SERVER_KEEP_ALIVE", struct.unpack("!H", b[1:3])[0], 3
elif id == PROP_RECEIVE_MAXIMUM:
return "RECEIVE_MAXIMUM", struct.unpack("!H", b[1:3])[0], 3
elif id == PROP_TOPIC_ALIAS_MAXIMUM:
return "TOPIC_ALIAS_MAXIMUM", struct.unpack("!H", b[1:3])[0], 3
elif id == PROP_TOPIC_ALIAS:
return "TOPIC_ALIAS", struct.unpack("!H", b[1:3])[0], 3
elif id == PROP_MESSAGE_EXPIRY_INTERVAL:
return "MESSAGE_EXPIRY_INTERVAL", struct.unpack("!B", b[1:5])[0], 5
elif id == PROP_SESSION_EXPIRY_INTERVAL:
return "SESSION_EXPIRY_INTERVAL", struct.unpack("!B", b[1:5])[0], 5
elif id == PROP_WILL_DELAY_INTERVAL:
return "WILL_DELAY_INTERVAL", struct.unpack("!B", b[1:5])[0], 5
elif id == PROP_MAXIMUM_PACKET_SIZE:
return "MAXIMUM_PACKET_SIZE", struct.unpack("!B", b[1:5])[0], 5
elif id == PROP_CONTENT_TYPE:
value, pack_len = unpack_string(b[1:])
return "CONTENT_TYPE", value, pack_len + 1
elif id == PROP_RESPONSE_TOPIC:
value, pack_len = unpack_string(b[1:])
return "RESPONSE_TOPIC", value, pack_len + 1
elif id == PROP_CORRELATION_DATA:
value, pack_len = unpack_string(b[1:])
return "CORRELATION_DATA", value, pack_len + 1
elif id == PROP_ASSIGNED_CLIENT_IDENTIFIER:
value, pack_len = unpack_string(b[1:])
return "ASSIGNED_CLIENT_IDENTIFIER", value, pack_len + 1
elif id == PROP_AUTHENTICATION_METHOD:
value, pack_len = unpack_string(b[1:])
return "AUTHENTICATION_METHOD", value, pack_len + 1
elif id == PROP_AUTHENTICATION_DATA:
value, pack_len = unpack_string(b[1:])
return "AUTHENTICATION_DATA", value, pack_len + 1
elif id == PROP_RESPONSE_INFO:
value, pack_len = unpack_string(b[1:])
return "RESPONSE_INFO", value, pack_len + 1
elif id == PROP_SERVER_REFERENCE:
value, pack_len = unpack_string(b[1:])
return "SERVER_REFERENCE", value, pack_len + 1
elif id == PROP_REASON_STRING:
value, pack_len = unpack_string(b[1:])
return "REASON_STRING", value, pack_len + 1
elif id == PROP_SUBSCRIPTION_IDENTIFIER:
value, pack_len = unpack_varint(b[1:])
return "REASON_STRING", value, pack_len + 1
elif id == PROP_USER_PROPERTY:
name, pack_name_len = unpack_string(b[1:])
value, pack_value_len = unpack_varint(b[1 + pack_name_len :])
return (
f"PROP_USER_PROPERTY:{name}",
value,
1 + pack_name_len + pack_value_len + pack_len,
)
else:
return f"<Unknown property ID={id}>", f"not decoded (len<={len(b)-1})", len(b)
def print_properties(b: bytes):
_, offset = unpack_varint(b)
props = []
while offset < len(b):
try:
key, value, prop_len = unpack_property(b[offset:])
offset += prop_len
props.append(f"{key}:{value}")
except struct.error:
props.append(f"decode error at offset {offset}: {b}")
break
return f"[{','.join(props)}]"
test/unit/broker/subs_stubs.c
+5
View File
@@ -108,6 +108,11 @@ int retain__init(void)
return MOSQ_ERR_SUCCESS;
}
void retain__expiry_check(struct mosquitto__retainhier **retainhier)
{
UNUSED(retainhier);
}
void retain__clean(struct mosquitto__retainhier **retainhier)
{
UNUSED(retainhier);