03-publish-b2c-disconnect-qos1.py 3.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384
  1. #!/usr/bin/env python3
  2. # Does an interrupted QoS 1 flow from broker to client get handled correctly?
  3. from mosq_test_helper import *
  4. def helper(port):
  5. connect_packet = mosq_test.gen_connect("test-helper", keepalive=60)
  6. connack_packet = mosq_test.gen_connack(rc=0)
  7. mid = 128
  8. publish_packet = mosq_test.gen_publish("qos1/disconnect/test", qos=1, mid=mid, payload="disconnect-message")
  9. puback_packet = mosq_test.gen_puback(mid)
  10. sock = mosq_test.do_client_connect(connect_packet, connack_packet, connack_error="helper connack", port=port)
  11. mosq_test.do_send_receive(sock, publish_packet, puback_packet, "helper puback")
  12. sock.close()
  13. def do_test(proto_ver):
  14. port = mosq_test.get_port()
  15. rc = 1
  16. mid = 3265
  17. keepalive = 60
  18. connect_packet = mosq_test.gen_connect("pub-qos1-disco-test", keepalive=keepalive, clean_session=False, proto_ver=proto_ver, session_expiry=60)
  19. connack1_packet = mosq_test.gen_connack(flags=0, rc=0, proto_ver=proto_ver)
  20. connack2_packet = mosq_test.gen_connack(flags=1, rc=0, proto_ver=proto_ver)
  21. subscribe_packet = mosq_test.gen_subscribe(mid, "qos1/disconnect/test", 1, proto_ver=proto_ver)
  22. suback_packet = mosq_test.gen_suback(mid, 1, proto_ver=proto_ver)
  23. mid = 1
  24. publish_packet = mosq_test.gen_publish("qos1/disconnect/test", qos=1, mid=mid, payload="disconnect-message", proto_ver=proto_ver)
  25. publish_dup_packet = mosq_test.gen_publish("qos1/disconnect/test", qos=1, mid=mid, payload="disconnect-message", dup=True, proto_ver=proto_ver)
  26. puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
  27. mid = 3266
  28. publish2_packet = mosq_test.gen_publish("qos1/outgoing", qos=1, mid=mid, payload="outgoing-message", proto_ver=proto_ver)
  29. puback2_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
  30. broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
  31. try:
  32. sock = mosq_test.do_client_connect(connect_packet, connack1_packet, port=port)
  33. mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
  34. helper(port)
  35. # Should have now received a publish command
  36. mosq_test.expect_packet(sock, "publish", publish_packet)
  37. # Send our outgoing message. When we disconnect the broker
  38. # should get rid of it and assume we're going to retry.
  39. sock.send(publish2_packet)
  40. sock.close()
  41. sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  42. sock.settimeout(60) # 60 seconds timeout is much longer than 5 seconds message retry.
  43. sock.connect(("localhost", port))
  44. mosq_test.do_send_receive(sock, connect_packet, connack2_packet, "connack")
  45. mosq_test.expect_packet(sock, "dup publish", publish_dup_packet)
  46. sock.send(puback_packet)
  47. rc = 0
  48. sock.close()
  49. except mosq_test.TestError:
  50. pass
  51. finally:
  52. broker.terminate()
  53. broker.wait()
  54. (stdo, stde) = broker.communicate()
  55. if rc:
  56. print(stde.decode('utf-8'))
  57. print("proto_ver=%d" % (proto_ver))
  58. exit(rc)
  59. do_test(proto_ver=4)
  60. do_test(proto_ver=5)
  61. exit(0)