Test and fix for message expiry not increasing send quota.
This commit is contained in:
parent
9c3a214a01
commit
042c2733f2
@ -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);
|
||||
|
93
test/broker/03-publish-qos1-max-inflight-expire.py
Executable file
93
test/broker/03-publish-qos1-max-inflight-expire.py
Executable file
@ -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)
|
@ -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
|
||||
|
@ -73,6 +73,7 @@ tests = [
|
||||
(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'),
|
||||
|
@ -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:
|
||||
|
Loading…
Reference in New Issue
Block a user