From 042c2733f25add2e3423a2d4bf83aff695e2d27c Mon Sep 17 00:00:00 2001 From: "Roger A. Light" Date: Mon, 2 Nov 2020 23:01:03 +0000 Subject: [PATCH] Test and fix for message expiry not increasing send quota. --- src/database.c | 4 + .../03-publish-qos1-max-inflight-expire.py | 93 +++++++++++++++++++ test/broker/Makefile | 1 + test/broker/test.py | 1 + test/mosq_test.py | 4 - 5 files changed, 99 insertions(+), 4 deletions(-) create mode 100755 test/broker/03-publish-qos1-max-inflight-expire.py diff --git a/src/database.c b/src/database.c index d911c64720..345c083e2c 100644 --- a/src/database.c +++ b/src/database.c @@ -954,6 +954,7 @@ int db__message_write_inflight_in(struct mosquitto_db *db, struct mosquitto *con if(now > tail->store->message_expiry_time){ /* Message is expired, must not send. */ db__message_remove(db, &context->msgs_in, tail); + util__decrement_send_quota(context); continue; } } @@ -1016,6 +1017,9 @@ static int db__message_write_inflight_out_single(struct mosquitto_db *db, struct if(now > msg->store->message_expiry_time){ /* Message is expired, must not send. */ db__message_remove(db, &context->msgs_out, msg); + if(msg->direction == mosq_md_out && msg->qos > 0){ + util__increment_send_quota(context); + } return MOSQ_ERR_SUCCESS; }else{ expiry_interval = (uint32_t)(msg->store->message_expiry_time - now); diff --git a/test/broker/03-publish-qos1-max-inflight-expire.py b/test/broker/03-publish-qos1-max-inflight-expire.py new file mode 100755 index 0000000000..54cb06a2af --- /dev/null +++ b/test/broker/03-publish-qos1-max-inflight-expire.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 + +# Test whether a PUBLISH to a topic with QoS 1 results in the correct packet flow for a subscriber. +# With max_inflight_messages set to 1 + +from mosq_test_helper import * + +def write_config(filename, port): + with open(filename, 'w') as f: + f.write("port %d\n" % (port)) + f.write("allow_anonymous true\n") + f.write("max_inflight_messages 1\n") + +def do_test(proto_ver): + rc = 1 + keepalive = 60 + + properties = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_SESSION_EXPIRY_INTERVAL, 1000) + sub_connect_packet = mosq_test.gen_connect("sub", keepalive=keepalive, properties=properties, proto_ver=proto_ver, clean_session=False) + + properties = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_TOPIC_ALIAS_MAXIMUM, 10) \ + + mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 1) + sub_connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver, properties=properties, property_helper=False) + sub_connack_packet2 = mosq_test.gen_connack(rc=0, flags=1, proto_ver=proto_ver, properties=properties, property_helper=False) + + mid = 1 + subscribe_packet = mosq_test.gen_subscribe(mid, "pub/qos1/test", 1, proto_ver=proto_ver) + suback_packet = mosq_test.gen_suback(mid, 1, proto_ver=proto_ver) + + connect_packet = mosq_test.gen_connect("pub-qos1-test", keepalive=keepalive, proto_ver=proto_ver) + properties = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_TOPIC_ALIAS_MAXIMUM, 10) \ + + mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 1) + connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver, properties=properties, property_helper=False) + + mid = 311 + properties = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_MESSAGE_EXPIRY_INTERVAL, 1) + publish_packet = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message", proto_ver=proto_ver, properties=properties) + puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver) + + mid = 1 + r_publish_packet = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message", proto_ver=proto_ver) + r_puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver) + + port = mosq_test.get_port() + conf_file = os.path.basename(__file__).replace('.py', '.conf') + write_config(conf_file, port) + broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port) + + try: + sub_sock = mosq_test.do_client_connect(sub_connect_packet, sub_connack_packet, port=port, timeout=10) + mosq_test.do_send_receive(sub_sock, subscribe_packet, suback_packet, "suback") + sub_sock.close() + + sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port, timeout=10) + mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback") + + time.sleep(2) + + sub_sock = mosq_test.do_client_connect(sub_connect_packet, sub_connack_packet2, port=port, timeout=10) + # This message has expired, so we don't expect it + #mosq_test.expect_packet(sub_sock, "publish 2", r_publish_packet) + #sub_sock.send(r_puback_packet) + + # + mid = 1 + s_publish_packet = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message2", proto_ver=proto_ver) + s_puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver) + mosq_test.do_send_receive(sock, s_publish_packet, s_puback_packet, "puback") + + mid = 2 + r_publish_packet = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message2", proto_ver=proto_ver) + r_puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver) + mosq_test.expect_packet(sub_sock, "publish 3", r_publish_packet) + sub_sock.send(r_puback_packet) + + rc = 0 + + sock.close() + except mosq_test.TestError: + pass + finally: + os.remove(conf_file) + broker.terminate() + broker.wait() + (stdo, stde) = broker.communicate() + if rc: + print(stde.decode('utf-8')) + print("proto_ver=%d" % (proto_ver)) + exit(rc) + + +do_test(proto_ver=5) +exit(0) diff --git a/test/broker/Makefile b/test/broker/Makefile index 4baa492f85..9797e942ac 100644 --- a/test/broker/Makefile +++ b/test/broker/Makefile @@ -91,6 +91,7 @@ test : test-compile 01 02 03 04 05 06 07 08 09 10 11 12 13 14 ./03-publish-dollar.py ./03-publish-invalid-utf8.py ./03-publish-long-topic.py + ./03-publish-qos1-max-inflight-expire.py ./03-publish-qos1-no-subscribers-v5.py ./03-publish-qos1-retain-disabled.py ./03-publish-qos1.py diff --git a/test/broker/test.py b/test/broker/test.py index d23c2d615d..85eb893b71 100755 --- a/test/broker/test.py +++ b/test/broker/test.py @@ -73,6 +73,7 @@ (1, './03-publish-dollar.py'), (1, './03-publish-invalid-utf8.py'), (1, './03-publish-long-topic.py'), + (1, './03-publish-qos1-max-inflight-expire.py'), (1, './03-publish-qos1-max-inflight.py'), (1, './03-publish-qos1-no-subscribers-v5.py'), (1, './03-publish-qos1-retain-disabled.py'), diff --git a/test/mosq_test.py b/test/mosq_test.py index f967c7acc9..6fc3ed4e87 100644 --- a/test/mosq_test.py +++ b/test/mosq_test.py @@ -100,14 +100,10 @@ def packet_matches(name, recvd, expected): print("Received: "+to_string(recvd)) except struct.error: print("Received (not decoded, len=%d): %s" % (len(recvd), recvd)) - for i in range(0, len(recvd)): - print('%c'%(recvd[i]),) try: print("Expected: "+to_string(expected)) except struct.error: print("Expected (not decoded, len=%d): %s" % (len(expected), expected)) - for i in range(0, len(expected)): - print('%c'%(expected[i]),) return False else: