03-publish-qos1-max-inflight-expire.py 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293
  1. #!/usr/bin/env python3
  2. # Test whether a PUBLISH to a topic with QoS 1 results in the correct packet flow for a subscriber.
  3. # With max_inflight_messages set to 1
  4. from mosq_test_helper import *
  5. def write_config(filename, port):
  6. with open(filename, 'w') as f:
  7. f.write("port %d\n" % (port))
  8. f.write("allow_anonymous true\n")
  9. f.write("max_inflight_messages 1\n")
  10. def do_test(proto_ver):
  11. rc = 1
  12. keepalive = 60
  13. properties = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_SESSION_EXPIRY_INTERVAL, 1000)
  14. sub_connect_packet = mosq_test.gen_connect("sub", keepalive=keepalive, properties=properties, proto_ver=proto_ver, clean_session=False)
  15. properties = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_TOPIC_ALIAS_MAXIMUM, 10) \
  16. + mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 1)
  17. sub_connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver, properties=properties, property_helper=False)
  18. sub_connack_packet2 = mosq_test.gen_connack(rc=0, flags=1, proto_ver=proto_ver, properties=properties, property_helper=False)
  19. mid = 1
  20. subscribe_packet = mosq_test.gen_subscribe(mid, "pub/qos1/test", 1, proto_ver=proto_ver)
  21. suback_packet = mosq_test.gen_suback(mid, 1, proto_ver=proto_ver)
  22. connect_packet = mosq_test.gen_connect("pub-qos1-test", keepalive=keepalive, proto_ver=proto_ver)
  23. properties = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_TOPIC_ALIAS_MAXIMUM, 10) \
  24. + mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 1)
  25. connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver, properties=properties, property_helper=False)
  26. mid = 311
  27. properties = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_MESSAGE_EXPIRY_INTERVAL, 1)
  28. publish_packet = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message", proto_ver=proto_ver, properties=properties)
  29. puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
  30. mid = 1
  31. r_publish_packet = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message", proto_ver=proto_ver)
  32. r_puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
  33. port = mosq_test.get_port()
  34. conf_file = os.path.basename(__file__).replace('.py', '.conf')
  35. write_config(conf_file, port)
  36. broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
  37. try:
  38. sub_sock = mosq_test.do_client_connect(sub_connect_packet, sub_connack_packet, port=port, timeout=10)
  39. mosq_test.do_send_receive(sub_sock, subscribe_packet, suback_packet, "suback")
  40. sub_sock.close()
  41. sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port, timeout=10)
  42. mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback")
  43. time.sleep(2)
  44. sub_sock = mosq_test.do_client_connect(sub_connect_packet, sub_connack_packet2, port=port, timeout=10)
  45. # This message has expired, so we don't expect it
  46. #mosq_test.expect_packet(sub_sock, "publish 2", r_publish_packet)
  47. #sub_sock.send(r_puback_packet)
  48. #
  49. mid = 1
  50. s_publish_packet = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message2", proto_ver=proto_ver)
  51. s_puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
  52. mosq_test.do_send_receive(sock, s_publish_packet, s_puback_packet, "puback")
  53. mid = 2
  54. r_publish_packet = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message2", proto_ver=proto_ver)
  55. r_puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
  56. mosq_test.expect_packet(sub_sock, "publish 3", r_publish_packet)
  57. sub_sock.send(r_puback_packet)
  58. rc = 0
  59. sock.close()
  60. except mosq_test.TestError:
  61. pass
  62. finally:
  63. os.remove(conf_file)
  64. broker.terminate()
  65. broker.wait()
  66. (stdo, stde) = broker.communicate()
  67. if rc:
  68. print(stde.decode('utf-8'))
  69. print("proto_ver=%d" % (proto_ver))
  70. exit(rc)
  71. do_test(proto_ver=5)
  72. exit(0)