diff --git a/plugins/persist-sqlite/base_msgs.c b/plugins/persist-sqlite/base_msgs.c index 99f8e2a2..1f9bd956 100644 --- a/plugins/persist-sqlite/base_msgs.c +++ b/plugins/persist-sqlite/base_msgs.c @@ -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; +} diff --git a/plugins/persist-sqlite/clients.c b/plugins/persist-sqlite/clients.c index a0d2d84a..5e3f2073 100644 --- a/plugins/persist-sqlite/clients.c +++ b/plugins/persist-sqlite/clients.c @@ -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; } + diff --git a/plugins/persist-sqlite/init.c b/plugins/persist-sqlite/init.c index 93ca8c2e..bb2c515c 100644 --- a/plugins/persist-sqlite/init.c +++ b/plugins/persist-sqlite/init.c @@ -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 " diff --git a/plugins/persist-sqlite/persist_sqlite.h b/plugins/persist-sqlite/persist_sqlite.h index ad629df2..6c0e7d88 100644 --- a/plugins/persist-sqlite/persist_sqlite.h +++ b/plugins/persist-sqlite/persist_sqlite.h @@ -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); diff --git a/src/database.c b/src/database.c index 05c3939a..fad8d08a 100644 --- a/src/database.c +++ b/src/database.c @@ -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); diff --git a/src/mosquitto.c b/src/mosquitto.c index 6b19ac40..9b08bfe3 100644 --- a/src/mosquitto.c +++ b/src/mosquitto.c @@ -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, diff --git a/src/mosquitto_broker_internal.h b/src/mosquitto_broker_internal.h index a543b066..b26db61f 100644 --- a/src/mosquitto_broker_internal.h +++ b/src/mosquitto_broker_internal.h @@ -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 * ============================================================ */ diff --git a/src/plugin_public.c b/src/plugin_public.c index 2541d31d..2afcc9ac 100644 --- a/src/plugin_public.c +++ b/src/plugin_public.c @@ -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: diff --git a/src/retain.c b/src/retain.c index 6ed18929..331581f4 100644 --- a/src/retain.c +++ b/src/retain.c @@ -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) } } + diff --git a/src/session_expiry.c b/src/session_expiry.c index c8697beb..03f5321c 100644 --- a/src/session_expiry.c +++ b/src/session_expiry.c @@ -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){ diff --git a/test/broker/15-persist-bridge-queue.py b/test/broker/15-persist-bridge-queue.py index 0598686a..0186cb5e 100755 --- a/test/broker/15-persist-bridge-queue.py +++ b/test/broker/15-persist-bridge-queue.py @@ -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={}, ) diff --git a/test/broker/15-persist-client-drop-expired-messages.py b/test/broker/15-persist-client-drop-expired-messages.py new file mode 100755 index 00000000..7e5231e5 --- /dev/null +++ b/test/broker/15-persist-client-drop-expired-messages.py @@ -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, +) diff --git a/test/broker/15-persist-client-expired-session.py b/test/broker/15-persist-client-expired-session.py new file mode 100755 index 00000000..c8826e09 --- /dev/null +++ b/test/broker/15-persist-client-expired-session.py @@ -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, +# ) + diff --git a/test/broker/15-persist-client-msg-modify-acl.py b/test/broker/15-persist-client-msg-modify-acl.py index 00722b69..76c34d36 100755 --- a/test/broker/15-persist-client-msg-modify-acl.py +++ b/test/broker/15-persist-client-msg-modify-acl.py @@ -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( diff --git a/test/broker/Makefile b/test/broker/Makefile index f93ee2c8..5b8ddbde 100644 --- a/test/broker/Makefile +++ b/test/broker/Makefile @@ -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 \ diff --git a/test/broker/mosq_test_helper.py b/test/broker/mosq_test_helper.py index aa5c1e6e..2d2450a4 100644 --- a/test/broker/mosq_test_helper.py +++ b/test/broker/mosq_test_helper.py @@ -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 diff --git a/test/broker/persist_module_helper.py b/test/broker/persist_module_helper.py new file mode 100644 index 00000000..e6a83e36 --- /dev/null +++ b/test/broker/persist_module_helper.py @@ -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, + ) diff --git a/test/broker/persist_sqlite.py b/test/broker/persist_sqlite.py index e0bd606a..e7888761 100755 --- a/test/broker/persist_sqlite.py +++ b/test/broker/persist_sqlite.py @@ -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() diff --git a/test/mosq_test.py b/test/mosq_test.py index 285b0e39..ebfe1af1 100644 --- a/test/mosq_test.py +++ b/test/mosq_test.py @@ -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)" diff --git a/test/mqtt5_props.py b/test/mqtt5_props.py index 9f6ad4bc..7ddd11c2 100644 --- a/test/mqtt5_props.py +++ b/test/mqtt5_props.py @@ -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"", 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)}]" diff --git a/test/unit/broker/subs_stubs.c b/test/unit/broker/subs_stubs.c index 32162345..14a89e7c 100644 --- a/test/unit/broker/subs_stubs.c +++ b/test/unit/broker/subs_stubs.c @@ -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);