Add MQTT v5 support to bridge tests.

test/broker/06-bridge-b2br-disconnect-qos1.py

@@ -4,7 +4,7 @@
 
 from mosq_test_helper import *
 
-def write_config(filename, port1, port2):
+def write_config(filename, port1, port2, protocol_version):
     with open(filename, 'w') as f:
         f.write("port %d\n" % (port2))
         f.write("\n")
@@ -13,85 +13,100 @@ def write_config(filename, port1, port2):
         f.write("topic bridge/# both 1\n")
         f.write("notifications false\n")
         f.write("restart_timeout 5\n")
+        f.write("bridge_protocol_version %s\n" %(protocol_version))
 
 (port1, port2) = mosq_test.get_port(2)
 conf_file = os.path.basename(__file__).replace('.py', '.conf')
-write_config(conf_file, port1, port2)
 
-rc = 1
-keepalive = 60
-client_id = socket.gethostname()+".bridge_sample"
-connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=128+4)
-connack_packet = mosq_test.gen_connack(rc=0)
 
-mid = 1
-subscribe_packet = mosq_test.gen_subscribe(mid, "bridge/#", 1)
-suback_packet = mosq_test.gen_suback(mid, 1)
+def do_test(proto_ver):
+    if proto_ver == 4:
+        bridge_protocol = "mqttv311"
+        proto_ver_connect = 128+4
+    else:
+        bridge_protocol = "mqttv50"
+        proto_ver_connect = 5
 
-mid = 2
-subscribe2_packet = mosq_test.gen_subscribe(mid, "bridge/#", 1)
-suback2_packet = mosq_test.gen_suback(mid, 1)
+    write_config(conf_file, port1, port2, bridge_protocol)
 
-mid = 3
-publish_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=1, mid=mid, payload="disconnect-message")
-publish_dup_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=1, mid=mid, payload="disconnect-message", dup=True)
-puback_packet = mosq_test.gen_puback(mid)
+    rc = 1
+    keepalive = 60
+    client_id = socket.gethostname()+".bridge_sample"
+    connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=proto_ver_connect)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
 
-mid = 20
-publish2_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=1, mid=mid, payload="disconnect-message")
-puback2_packet = mosq_test.gen_puback(mid)
+    mid = 1
+    subscribe_packet = mosq_test.gen_subscribe(mid, "bridge/#", 1, proto_ver=proto_ver)
+    suback_packet = mosq_test.gen_suback(mid, 1, proto_ver=proto_ver)
 
-ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-ssock.settimeout(40)
-ssock.bind(('', port1))
-ssock.listen(5)
+    mid = 2
+    subscribe2_packet = mosq_test.gen_subscribe(mid, "bridge/#", 1, proto_ver=proto_ver)
+    suback2_packet = mosq_test.gen_suback(mid, 1, proto_ver=proto_ver)
 
-broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
+    mid = 3
+    publish_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=1, mid=mid, payload="disconnect-message", proto_ver=proto_ver)
+    publish_dup_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=1, mid=mid, payload="disconnect-message", dup=True, proto_ver=proto_ver)
+    puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
 
-try:
-    (bridge, address) = ssock.accept()
-    bridge.settimeout(20)
+    mid = 20
+    publish2_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=1, mid=mid, payload="disconnect-message", proto_ver=proto_ver)
+    puback2_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
 
-    if mosq_test.expect_packet(bridge, "connect", connect_packet):
-        bridge.send(connack_packet)
+    ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    ssock.settimeout(40)
+    ssock.bind(('', port1))
+    ssock.listen(5)
 
-        if mosq_test.expect_packet(bridge, "subscribe", subscribe_packet):
-            bridge.send(suback_packet)
+    try:
+        broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
 
-            bridge.send(publish_packet)
-            # Bridge doesn't have time to respond but should expect us to retry
-            # and so remove PUBACK.
-            bridge.close()
+        (bridge, address) = ssock.accept()
+        bridge.settimeout(20)
 
-            (bridge, address) = ssock.accept()
-            bridge.settimeout(20)
+        if mosq_test.expect_packet(bridge, "connect", connect_packet):
+            bridge.send(connack_packet)
 
-            if mosq_test.expect_packet(bridge, "connect", connect_packet):
-                bridge.send(connack_packet)
+            if mosq_test.expect_packet(bridge, "subscribe", subscribe_packet):
+                bridge.send(suback_packet)
 
-                if mosq_test.expect_packet(bridge, "2nd subscribe", subscribe2_packet):
-                    bridge.send(suback2_packet)
+                bridge.send(publish_packet)
+                # Bridge doesn't have time to respond but should expect us to retry
+                # and so remove PUBACK.
+                bridge.close()
 
-                    # Send a different publish message to make sure the response isn't to the old one.
-                    bridge.send(publish2_packet)
-                    if mosq_test.expect_packet(bridge, "puback", puback2_packet):
-                        rc = 0
+                (bridge, address) = ssock.accept()
+                bridge.settimeout(20)
+
+                if mosq_test.expect_packet(bridge, "connect", connect_packet):
+                    bridge.send(connack_packet)
+
+                    if mosq_test.expect_packet(bridge, "2nd subscribe", subscribe2_packet):
+                        bridge.send(suback2_packet)
+
+                        # Send a different publish message to make sure the response isn't to the old one.
+                        bridge.send(publish2_packet)
+                        if mosq_test.expect_packet(bridge, "puback", puback2_packet):
+                            rc = 0
 
-    bridge.close()
-finally:
-    os.remove(conf_file)
-    try:
         bridge.close()
-    except NameError:
-        pass
+    finally:
+        os.remove(conf_file)
+        try:
+            bridge.close()
+        except NameError:
+            pass
+
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        ssock.close()
+        if rc:
+            print(stde.decode('utf-8'))
+            exit(rc)
 
-    broker.terminate()
-    broker.wait()
-    (stdo, stde) = broker.communicate()
-    if rc:
-        print(stde.decode('utf-8'))
-    ssock.close()
+do_test(proto_ver=4)
+do_test(proto_ver=5)
 
-exit(rc)
+exit(0)
 

test/broker/06-bridge-b2br-disconnect-qos2.py

@@ -4,7 +4,7 @@
 
 from mosq_test_helper import *
 
-def write_config(filename, port1, port2):
+def write_config(filename, port1, port2, protocol_version):
     with open(filename, 'w') as f:
         f.write("port %d\n" % (port2))
         f.write("\n")
@@ -13,102 +13,116 @@ def write_config(filename, port1, port2):
         f.write("topic bridge/# both 2\n")
         f.write("notifications false\n")
         f.write("restart_timeout 5\n")
+        f.write("bridge_protocol_version %s\n" % (protocol_version))
 
 (port1, port2) = mosq_test.get_port(2)
 conf_file = os.path.basename(__file__).replace('.py', '.conf')
-write_config(conf_file, port1, port2)
 
-rc = 1
-keepalive = 60
-client_id = socket.gethostname()+".bridge_sample"
-connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=128+4)
-connack_packet = mosq_test.gen_connack(rc=0)
+def do_test(proto_ver):
+    if proto_ver == 4:
+        bridge_protocol = "mqttv311"
+        proto_ver_connect = 128+4
+    else:
+        bridge_protocol = "mqttv50"
+        proto_ver_connect = 5
+        
+    write_config(conf_file, port1, port2, bridge_protocol)
+
+    rc = 1
+    keepalive = 60
+    client_id = socket.gethostname()+".bridge_sample"
+    connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=proto_ver_connect)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+
+    mid = 1
+    subscribe_packet = mosq_test.gen_subscribe(mid, "bridge/#", 2, proto_ver=proto_ver)
+    suback_packet = mosq_test.gen_suback(mid, 2, proto_ver=proto_ver)
+
+    mid = 2
+    subscribe2_packet = mosq_test.gen_subscribe(mid, "bridge/#", 2, proto_ver=proto_ver)
+    suback2_packet = mosq_test.gen_suback(mid, 2, proto_ver=proto_ver)
+
+    mid = 3
+    subscribe3_packet = mosq_test.gen_subscribe(mid, "bridge/#", 2, proto_ver=proto_ver)
+    suback3_packet = mosq_test.gen_suback(mid, 2, proto_ver=proto_ver)
+
+    mid = 5
+    publish_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=2, mid=mid, payload="disconnect-message", proto_ver=proto_ver)
+    publish_dup_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=2, mid=mid, payload="disconnect-message", dup=True, proto_ver=proto_ver)
+    pubrec_packet = mosq_test.gen_pubrec(mid, proto_ver=proto_ver)
+    pubrel_packet = mosq_test.gen_pubrel(mid, proto_ver=proto_ver)
+    pubcomp_packet = mosq_test.gen_pubcomp(mid, proto_ver=proto_ver)
+
+    ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    ssock.settimeout(40)
+    ssock.bind(('', port1))
+    ssock.listen(5)
 
-mid = 1
-subscribe_packet = mosq_test.gen_subscribe(mid, "bridge/#", 2)
-suback_packet = mosq_test.gen_suback(mid, 2)
-
-mid = 2
-subscribe2_packet = mosq_test.gen_subscribe(mid, "bridge/#", 2)
-suback2_packet = mosq_test.gen_suback(mid, 2)
-
-mid = 3
-subscribe3_packet = mosq_test.gen_subscribe(mid, "bridge/#", 2)
-suback3_packet = mosq_test.gen_suback(mid, 2)
-
-mid = 5
-publish_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=2, mid=mid, payload="disconnect-message")
-publish_dup_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=2, mid=mid, payload="disconnect-message", dup=True)
-pubrec_packet = mosq_test.gen_pubrec(mid)
-pubrel_packet = mosq_test.gen_pubrel(mid)
-pubcomp_packet = mosq_test.gen_pubcomp(mid)
-
-ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-ssock.settimeout(40)
-ssock.bind(('', port1))
-ssock.listen(5)
+    try:
+        broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
+        (bridge, address) = ssock.accept()
+        bridge.settimeout(20)
 
-broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
+        if mosq_test.expect_packet(bridge, "connect", connect_packet):
+            bridge.send(connack_packet)
 
-try:
-    (bridge, address) = ssock.accept()
-    bridge.settimeout(20)
+            if mosq_test.expect_packet(bridge, "subscribe", subscribe_packet):
+                bridge.send(suback_packet)
 
-    if mosq_test.expect_packet(bridge, "connect", connect_packet):
-        bridge.send(connack_packet)
+                bridge.send(publish_packet)
+                bridge.close()
 
-        if mosq_test.expect_packet(bridge, "subscribe", subscribe_packet):
-            bridge.send(suback_packet)
+                (bridge, address) = ssock.accept()
+                bridge.settimeout(20)
 
-            bridge.send(publish_packet)
-            bridge.close()
+                if mosq_test.expect_packet(bridge, "connect", connect_packet):
+                    bridge.send(connack_packet)
 
-            (bridge, address) = ssock.accept()
-            bridge.settimeout(20)
+                    if mosq_test.expect_packet(bridge, "2nd subscribe", subscribe2_packet):
+                        bridge.send(suback2_packet)
+                        bridge.send(publish_dup_packet)
 
-            if mosq_test.expect_packet(bridge, "connect", connect_packet):
-                bridge.send(connack_packet)
+                        if mosq_test.expect_packet(bridge, "pubrec", pubrec_packet):
+                            bridge.send(pubrel_packet)
+                            bridge.close()
 
-                if mosq_test.expect_packet(bridge, "2nd subscribe", subscribe2_packet):
-                    bridge.send(suback2_packet)
-                    bridge.send(publish_dup_packet)
+                            (bridge, address) = ssock.accept()
+                            bridge.settimeout(20)
 
-                    if mosq_test.expect_packet(bridge, "pubrec", pubrec_packet):
-                        bridge.send(pubrel_packet)
-                        bridge.close()
+                            if mosq_test.expect_packet(bridge, "connect", connect_packet):
+                                bridge.send(connack_packet)
 
-                        (bridge, address) = ssock.accept()
-                        bridge.settimeout(20)
+                                if mosq_test.expect_packet(bridge, "3rd subscribe", subscribe3_packet):
+                                    bridge.send(suback3_packet)
 
-                        if mosq_test.expect_packet(bridge, "connect", connect_packet):
-                            bridge.send(connack_packet)
+                                    bridge.send(publish_dup_packet)
 
-                            if mosq_test.expect_packet(bridge, "3rd subscribe", subscribe3_packet):
-                                bridge.send(suback3_packet)
+                                    if mosq_test.expect_packet(bridge, "2nd pubrec", pubrec_packet):
+                                        bridge.send(pubrel_packet)
 
-                                bridge.send(publish_dup_packet)
+                                        if mosq_test.expect_packet(bridge, "pubcomp", pubcomp_packet):
+                                            rc = 0
 
-                                if mosq_test.expect_packet(bridge, "2nd pubrec", pubrec_packet):
-                                    bridge.send(pubrel_packet)
+        bridge.close()
+    finally:
+        os.remove(conf_file)
+        try:
+            bridge.close()
+        except NameError:
+            pass
 
-                                    if mosq_test.expect_packet(bridge, "pubcomp", pubcomp_packet):
-                                        rc = 0
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        ssock.close()
+        if rc:
+            print(stde.decode('utf-8'))
+            exit(rc)
 
-    bridge.close()
-finally:
-    os.remove(conf_file)
-    try:
-        bridge.close()
-    except NameError:
-        pass
 
-    broker.terminate()
-    broker.wait()
-    (stdo, stde) = broker.communicate()
-    if rc:
-        print(stde.decode('utf-8'))
-    ssock.close()
+do_test(proto_ver=4)
+do_test(proto_ver=5)
 
-exit(rc)
+exit(0)
 

test/broker/06-bridge-b2br-late-connection-retain.py

@@ -11,7 +11,7 @@ def write_config1(filename, persistence_file, port1, port2):
         f.write("persistence true\n")
         f.write("persistence_file %s\n" % (persistence_file))
 
-def write_config2(filename, persistence_file, port1, port2):
+def write_config2(filename, persistence_file, port1, port2, protocol_version):
     with open(filename, 'w') as f:
         f.write("port %d\n" % (port2))
         f.write("\n")
@@ -20,6 +20,7 @@ def write_config2(filename, persistence_file, port1, port2):
         f.write("topic bridge/# out 1\n")
         f.write("notifications false\n")
         f.write("bridge_attempt_unsubscribe false\n")
+        f.write("bridge_protocol_version %s\n" % (protocol_version))
         f.write("persistence true\n")
         f.write("persistence_file %s\n" % (persistence_file))
 
@@ -27,69 +28,85 @@ def write_config2(filename, persistence_file, port1, port2):
 conf_file = os.path.basename(__file__).replace('.py', '.conf')
 persistence_file = os.path.basename(__file__).replace('.py', '.db')
 
-rc = 1
-keepalive = 60
-client_id = socket.gethostname()+".bridge_sample"
-connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=128+4)
-connack_packet = mosq_test.gen_connack(rc=0)
-
-c_connect_packet = mosq_test.gen_connect("client", keepalive=keepalive)
-c_connack_packet = mosq_test.gen_connack(rc=0)
-
-mid = 1
-publish_packet = mosq_test.gen_publish("bridge/test", qos=1, mid=mid, payload="message", retain=True)
-puback_packet = mosq_test.gen_puback(mid)
-
-ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-ssock.settimeout(40)
-ssock.bind(('', port1))
-ssock.listen(5)
-
-write_config1(conf_file, persistence_file, port1, port2)
-broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
-
-try:
-    client = mosq_test.do_client_connect(c_connect_packet, c_connack_packet, timeout=20, port=port2)
-    mosq_test.do_send_receive(client, publish_packet, puback_packet, "puback")
-    client.close()
-
-    broker.terminate()
-    broker.wait()
-
-    # Restart, with retained message in place
-    write_config2(conf_file, persistence_file, port1, port2)
-    broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
-
-    (bridge, address) = ssock.accept()
-    bridge.settimeout(20)
-
-    if mosq_test.expect_packet(bridge, "connect", connect_packet):
-        bridge.send(connack_packet)
-
-        if mosq_test.expect_packet(bridge, "publish", publish_packet):
-            bridge.send(puback_packet)
-            # Guard against multiple retained messages of the same type by
-            # sending a pingreq to give us something to expect back. If we get
-            # a publish, it's a fail.
-            mosq_test.do_ping(bridge)
-            rc = 0
-
-    bridge.close()
-finally:
-    os.remove(conf_file)
+def do_test(proto_ver):
+    if proto_ver == 4:
+        bridge_protocol = "mqttv311"
+        proto_ver_connect = 128+4
+    else:
+        bridge_protocol = "mqttv50"
+        proto_ver_connect = 5
+
+    rc = 1
+    keepalive = 60
+    client_id = socket.gethostname()+".bridge_sample"
+    connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=proto_ver_connect)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+
+    c_connect_packet = mosq_test.gen_connect("client", keepalive=keepalive, proto_ver=proto_ver)
+    c_connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+
+    mid = 1
+    publish_packet = mosq_test.gen_publish("bridge/test", qos=1, mid=mid, payload="message", retain=True, proto_ver=proto_ver)
+
+    if proto_ver == 5:
+        puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver, reason_code=16)
+    else:
+        puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
+
+    ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    ssock.settimeout(40)
+    ssock.bind(('', port1))
+    ssock.listen(5)
+
+    write_config1(conf_file, persistence_file, port1, port2)
+
     try:
-        bridge.close()
-    except NameError:
-        pass
+        broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
+        client = mosq_test.do_client_connect(c_connect_packet, c_connack_packet, timeout=20, port=port2)
+        mosq_test.do_send_receive(client, publish_packet, puback_packet, "puback")
+        client.close()
+
+        broker.terminate()
+        broker.wait()
 
-    broker.terminate()
-    broker.wait()
-    (stdo, stde) = broker.communicate()
-    os.remove(persistence_file)
-    if rc:
-        print(stde.decode('utf-8'))
-    ssock.close()
+        # Restart, with retained message in place
+        write_config2(conf_file, persistence_file, port1, port2, bridge_protocol)
+        broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
 
-exit(rc)
+        (bridge, address) = ssock.accept()
+        bridge.settimeout(20)
 
+        if mosq_test.expect_packet(bridge, "connect", connect_packet):
+            bridge.send(connack_packet)
+
+            if mosq_test.expect_packet(bridge, "publish", publish_packet):
+                bridge.send(puback_packet)
+                # Guard against multiple retained messages of the same type by
+                # sending a pingreq to give us something to expect back. If we get
+                # a publish, it's a fail.
+                mosq_test.do_ping(bridge)
+                rc = 0
+
+        bridge.close()
+    finally:
+        os.remove(conf_file)
+        try:
+            bridge.close()
+        except NameError:
+            pass
+
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        os.remove(persistence_file)
+        ssock.close()
+        if rc:
+            print(stde.decode('utf-8'))
+            exit(rc)
+
+
+do_test(proto_ver=4)
+do_test(proto_ver=5)
+
+exit(0)

test/broker/06-bridge-b2br-late-connection.py

@@ -4,7 +4,7 @@
 
 from mosq_test_helper import *
 
-def write_config(filename, port1, port2):
+def write_config(filename, port1, port2, protocol_version):
     with open(filename, 'w') as f:
         f.write("port %d\n" % (port2))
         f.write("\n")
@@ -13,61 +13,75 @@ def write_config(filename, port1, port2):
         f.write("topic bridge/# out 1\n")
         f.write("notifications false\n")
         f.write("bridge_attempt_unsubscribe false\n")
+        f.write("bridge_protocol_version %s\n" % (protocol_version))
 
 (port1, port2) = mosq_test.get_port(2)
 conf_file = os.path.basename(__file__).replace('.py', '.conf')
-write_config(conf_file, port1, port2)
 
-rc = 1
-keepalive = 60
-client_id = socket.gethostname()+".bridge_sample"
-connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=128+4)
-connack_packet = mosq_test.gen_connack(rc=0)
+def do_test(proto_ver):
+    if proto_ver == 4:
+        bridge_protocol = "mqttv311"
+        proto_ver_connect = 128+4
+    else:
+        bridge_protocol = "mqttv50"
+        proto_ver_connect = 5
 
-c_connect_packet = mosq_test.gen_connect("client", keepalive=keepalive)
-c_connack_packet = mosq_test.gen_connack(rc=0)
+    write_config(conf_file, port1, port2, bridge_protocol)
 
-mid = 1
-publish_packet = mosq_test.gen_publish("bridge/test", qos=1, mid=mid, payload="message")
-puback_packet = mosq_test.gen_puback(mid)
+    rc = 1
+    keepalive = 60
+    client_id = socket.gethostname()+".bridge_sample"
+    connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=proto_ver_connect)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
 
-ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-ssock.settimeout(40)
-ssock.bind(('', port1))
-ssock.listen(5)
+    c_connect_packet = mosq_test.gen_connect("client", keepalive=keepalive, proto_ver=proto_ver)
+    c_connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
 
-broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
+    mid = 1
+    publish_packet = mosq_test.gen_publish("bridge/test", qos=1, mid=mid, payload="message", proto_ver=proto_ver)
+    puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
 
-try:
-    (bridge, address) = ssock.accept()
-    bridge.settimeout(20)
+    ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    ssock.settimeout(40)
+    ssock.bind(('', port1))
+    ssock.listen(5)
 
-    client = mosq_test.do_client_connect(c_connect_packet, c_connack_packet, timeout=20, port=port2)
-    mosq_test.do_send_receive(client, publish_packet, puback_packet, "puback")
-    client.close()
-    # We've now sent a message to the broker that should be delivered to us via the bridge
+    broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
 
-    if mosq_test.expect_packet(bridge, "connect", connect_packet):
-        bridge.send(connack_packet)
+    try:
+        (bridge, address) = ssock.accept()
+        bridge.settimeout(20)
 
-        if mosq_test.expect_packet(bridge, "publish", publish_packet):
-            rc = 0
+        client = mosq_test.do_client_connect(c_connect_packet, c_connack_packet, timeout=20, port=port2)
+        mosq_test.do_send_receive(client, publish_packet, puback_packet, "puback")
+        client.close()
+        # We've now sent a message to the broker that should be delivered to us via the bridge
 
-    bridge.close()
-finally:
-    os.remove(conf_file)
-    try:
-        bridge.close()
-    except NameError:
-        pass
+        if mosq_test.expect_packet(bridge, "connect", connect_packet):
+            bridge.send(connack_packet)
 
-    broker.terminate()
-    broker.wait()
-    (stdo, stde) = broker.communicate()
-    if rc:
-        print(stde.decode('utf-8'))
-    ssock.close()
+            if mosq_test.expect_packet(bridge, "publish", publish_packet):
+                rc = 0
 
-exit(rc)
+        bridge.close()
+    finally:
+        os.remove(conf_file)
+        try:
+            bridge.close()
+        except NameError:
+            pass
+
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        ssock.close()
+        if rc:
+            print(stde.decode('utf-8'))
+            exit(rc)
+
+do_test(proto_ver=4)
+do_test(proto_ver=5)
+
+exit(0)
 

test/broker/06-bridge-b2br-remapping.py

@@ -4,7 +4,7 @@
 
 from mosq_test_helper import *
 
-def write_config(filename, port1, port2):
+def write_config(filename, port1, port2, protocol_version):
     with open(filename, 'w') as f:
         f.write("port %d\n" % (port2))
         f.write("\n")
@@ -18,30 +18,18 @@ def write_config(filename, port1, port2):
         f.write("topic clients/total in 0 test/mosquitto/org $SYS/broker/\n")
         f.write("notifications false\n")
         f.write("restart_timeout 5\n")
+        f.write("bridge_protocol_version %s\n" % (protocol_version))
 
 (port1, port2) = mosq_test.get_port(2)
 conf_file = os.path.basename(__file__).replace('.py', '.conf')
-write_config(conf_file, port1, port2)
 
-rc = 1
-keepalive = 60
-client_id = socket.gethostname()+".bridge_sample"
-connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=128+4)
-connack_packet = mosq_test.gen_connack(rc=0)
+connect_packet = None
+connack_packet = None
 
-client_connect_packet = mosq_test.gen_connect("pub-test", keepalive=keepalive)
-client_connack_packet = mosq_test.gen_connack(rc=0)
 
-ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-ssock.settimeout(4)
-ssock.bind(('', port1))
-ssock.listen(5)
+def inner_test(bridge, sock, proto_ver):
+    global connect_packet, connack_packet
 
-broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
-
-
-def test(bridge, sock):
     if not mosq_test.expect_packet(bridge, "connect", connect_packet):
         return 1
     bridge.send(connack_packet)
@@ -56,15 +44,15 @@ def test(bridge, sock):
     ]
     for pattern in ("remote/topic/#", "remote2/topic/prefix/#", "remote3/topic/+/value"):
         mid += 1
-        subscribe_packet = mosq_test.gen_subscribe(mid, pattern, 0)
-        suback_packet = mosq_test.gen_suback(mid, 0)
+        subscribe_packet = mosq_test.gen_subscribe(mid, pattern, 0, proto_ver=proto_ver)
+        suback_packet = mosq_test.gen_suback(mid, 0, proto_ver=proto_ver)
         if not mosq_test.expect_packet(bridge, "subscribe", subscribe_packet):
             return 1
         bridge.send(suback_packet)
 
     mid += 1
-    subscribe_packet = mosq_test.gen_subscribe(mid, "#", 0)
-    suback_packet = mosq_test.gen_suback(mid, 0)
+    subscribe_packet = mosq_test.gen_subscribe(mid, "#", 0, proto_ver=proto_ver)
+    suback_packet = mosq_test.gen_suback(mid, 0, proto_ver=proto_ver)
     sock.send(subscribe_packet)
     if not mosq_test.expect_packet(sock, "suback", suback_packet):
         return 1
@@ -84,11 +72,9 @@ def test(bridge, sock):
     for (local_topic, remote_topic) in cases:
         mid += 1
         remote_publish_packet = mosq_test.gen_publish(
-            remote_topic, qos=0, mid=mid, payload=''
-        )
+            remote_topic, qos=0, mid=mid, payload='', proto_ver=proto_ver)
         local_publish_packet = mosq_test.gen_publish(
-            local_topic, qos=0, mid=mid, payload=''
-        )
+            local_topic, qos=0, mid=mid, payload='', proto_ver=proto_ver)
 
         bridge.send(remote_publish_packet)
         match = mosq_test.expect_packet(sock, "publish", local_publish_packet)
@@ -99,31 +85,67 @@ def test(bridge, sock):
             return 1
     return 0
 
-try:
-    (bridge, address) = ssock.accept()
-    bridge.settimeout(2)
 
-    sock = mosq_test.do_client_connect(
-        client_connect_packet, client_connack_packet,
-        port=port2,
-    )
+def do_test(proto_ver):
+    global connect_packet, connack_packet
+
+    if proto_ver == 4:
+        bridge_protocol = "mqttv311"
+        proto_ver_connect = 128+4
+    else:
+        bridge_protocol = "mqttv50"
+        proto_ver_connect = 5
+
+    write_config(conf_file, port1, port2, bridge_protocol)
+
+    rc = 1
+    keepalive = 60
 
-    rc = test(bridge, sock)
+    client_id = socket.gethostname()+".bridge_sample"
 
-    sock.close()
-    bridge.close()
-finally:
-    os.remove(conf_file)
+    connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=proto_ver_connect)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+
+    client_connect_packet = mosq_test.gen_connect("pub-test", keepalive=keepalive, proto_ver=proto_ver)
+    client_connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+
+    ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    ssock.settimeout(4)
+    ssock.bind(('', port1))
+    ssock.listen(5)
+
+    broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
     try:
-        bridge.close()
-    except NameError:
-        pass
+        (bridge, address) = ssock.accept()
+        bridge.settimeout(2)
 
-    broker.terminate()
-    broker.wait()
-    (stdo, stde) = broker.communicate()
-    if rc:
-        print(stde.decode('utf-8'))
-    ssock.close()
+        sock = mosq_test.do_client_connect(
+            client_connect_packet, client_connack_packet,
+            port=port2,
+        )
+
+        rc = inner_test(bridge, sock, proto_ver)
 
-exit(rc)
+        sock.close()
+        bridge.close()
+    finally:
+        os.remove(conf_file)
+        try:
+            bridge.close()
+        except NameError:
+            pass
+
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        ssock.close()
+        if rc:
+            print(stde.decode('utf-8'))
+            exit(rc)
+
+
+do_test(proto_ver=4)
+do_test(proto_ver=5)
+
+exit(0)

test/broker/06-bridge-br2b-disconnect-qos1-helper.py

@@ -1,24 +0,0 @@
-#!/usr/bin/env python3
-
-from mosq_test_helper import *
-
-port = mosq_test.get_port()
-
-rc = 1
-keepalive = 60
-connect_packet = mosq_test.gen_connect("test-helper", keepalive=keepalive)
-connack_packet = mosq_test.gen_connack(rc=0)
-
-mid = 128
-publish_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=1, mid=mid, payload="disconnect-message")
-puback_packet = mosq_test.gen_puback(mid)
-
-sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port, connack_error="helper connack")
-mosq_test.do_send_receive(sock, publish_packet, puback_packet, "helper puback")
-
-rc = 0
-
-sock.close()
-
-exit(rc)
-

test/broker/06-bridge-br2b-disconnect-qos1.py

@@ -4,7 +4,7 @@
 
 from mosq_test_helper import *
 
-def write_config(filename, port1, port2):
+def write_config(filename, port1, port2, protocol_version):
     with open(filename, 'w') as f:
         f.write("port %d\n" % (port2))
         f.write("\n")
@@ -13,82 +13,104 @@ def write_config(filename, port1, port2):
         f.write("topic bridge/# both 1\n")
         f.write("notifications false\n")
         f.write("restart_timeout 5\n")
+        f.write("bridge_protocol_version %s\n" % (protocol_version))
 
 (port1, port2) = mosq_test.get_port(2)
 conf_file = os.path.basename(__file__).replace('.py', '.conf')
-write_config(conf_file, port1, port2)
 
-rc = 1
-keepalive = 60
-client_id = socket.gethostname()+".bridge_sample"
-connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=128+4)
-connack_packet = mosq_test.gen_connack(rc=0)
 
-mid = 1
-subscribe_packet = mosq_test.gen_subscribe(mid, "bridge/#", 1)
-suback_packet = mosq_test.gen_suback(mid, 1)
+def do_test(proto_ver):
+    if proto_ver == 4:
+        bridge_protocol = "mqttv311"
+        proto_ver_connect = 128+4
+    else:
+        bridge_protocol = "mqttv50"
+        proto_ver_connect = 5
 
-mid = 3
-subscribe2_packet = mosq_test.gen_subscribe(mid, "bridge/#", 1)
-suback2_packet = mosq_test.gen_suback(mid, 1)
+    write_config(conf_file, port1, port2, bridge_protocol)
 
-mid = 2
-publish_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=1, mid=mid, payload="disconnect-message")
-publish_dup_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=1, mid=mid, payload="disconnect-message", dup=True)
-puback_packet = mosq_test.gen_puback(mid)
+    rc = 1
+    keepalive = 60
+    client_id = socket.gethostname()+".bridge_sample"
+    connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=proto_ver_connect)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
 
-ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-ssock.settimeout(40)
-ssock.bind(('', port1))
-ssock.listen(5)
+    mid = 1
+    subscribe_packet = mosq_test.gen_subscribe(mid, "bridge/#", 1, proto_ver=proto_ver)
+    suback_packet = mosq_test.gen_suback(mid, 1, proto_ver=proto_ver)
 
-broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
+    mid = 3
+    subscribe2_packet = mosq_test.gen_subscribe(mid, "bridge/#", 1, proto_ver=proto_ver)
+    suback2_packet = mosq_test.gen_suback(mid, 1, proto_ver=proto_ver)
 
-try:
-    (bridge, address) = ssock.accept()
-    bridge.settimeout(20)
+    mid = 2
+    publish_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=1, mid=mid, payload="disconnect-message", proto_ver=proto_ver)
+    publish_dup_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=1, mid=mid, payload="disconnect-message", dup=True, proto_ver=proto_ver)
+    puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
 
-    if mosq_test.expect_packet(bridge, "connect", connect_packet):
-        bridge.send(connack_packet)
+    ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    ssock.settimeout(40)
+    ssock.bind(('', port1))
+    ssock.listen(5)
 
-        if mosq_test.expect_packet(bridge, "subscribe", subscribe_packet):
-            bridge.send(suback_packet)
+    broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
 
-            pub = subprocess.Popen(['./06-bridge-br2b-disconnect-qos1-helper.py', str(port2)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-            if pub.wait():
-                exit(1)
-            (stdo, stde) = pub.communicate()
+    #try:
+    if True:
+        (bridge, address) = ssock.accept()
+        bridge.settimeout(20)
 
-            if mosq_test.expect_packet(bridge, "publish", publish_packet):
-                bridge.close()
+        if mosq_test.expect_packet(bridge, "connect", connect_packet):
+            bridge.send(connack_packet)
 
-                (bridge, address) = ssock.accept()
-                bridge.settimeout(20)
+            if mosq_test.expect_packet(bridge, "subscribe", subscribe_packet):
+                bridge.send(suback_packet)
 
-                if mosq_test.expect_packet(bridge, "2nd connect", connect_packet):
-                    bridge.send(connack_packet)
+                # Helper
+                helper_connect_packet = mosq_test.gen_connect("test-helper", keepalive=keepalive, proto_ver=proto_ver)
+                helper_connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+                mid = 128
+                helper_publish_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=1, mid=mid, payload="disconnect-message", proto_ver=proto_ver)
+                helper_puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
+                helper_sock = mosq_test.do_client_connect(helper_connect_packet, helper_connack_packet, port=port2, connack_error="helper connack")
+                mosq_test.do_send_receive(helper_sock, publish_packet, puback_packet, "helper puback")
+                helper_sock.close()
+                # End helper
 
-                    if mosq_test.expect_packet(bridge, "2nd subscribe", subscribe2_packet):
-                        bridge.send(suback2_packet)
+                if mosq_test.expect_packet(bridge, "publish", publish_packet):
+                    bridge.close()
 
-                        if mosq_test.expect_packet(bridge, "2nd publish", publish_dup_packet):
-                            rc = 0
+                    (bridge, address) = ssock.accept()
+                    bridge.settimeout(20)
 
-    bridge.close()
-finally:
-    os.remove(conf_file)
-    try:
-        bridge.close()
-    except NameError:
-        pass
+                    if mosq_test.expect_packet(bridge, "2nd connect", connect_packet):
+                        bridge.send(connack_packet)
 
-    broker.terminate()
-    broker.wait()
-    (stdo, stde) = broker.communicate()
-    if rc:
-        print(stde.decode('utf-8'))
-    ssock.close()
+                        if mosq_test.expect_packet(bridge, "2nd subscribe", subscribe2_packet):
+                            bridge.send(suback2_packet)
 
-exit(rc)
+                            if mosq_test.expect_packet(bridge, "2nd publish", publish_dup_packet):
+                                rc = 0
 
+        bridge.close()
+    #finally:
+        os.remove(conf_file)
+        try:
+            bridge.close()
+        except NameError:
+            pass
+
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        ssock.close()
+        if rc:
+            print(stde.decode('utf-8'))
+            exit(rc)
+
+
+do_test(proto_ver=4)
+do_test(proto_ver=5)
+
+exit(0)

test/broker/06-bridge-br2b-disconnect-qos2-helper.py

@@ -1,30 +0,0 @@
-#!/usr/bin/env python3
-
-from mosq_test_helper import *
-
-port = mosq_test.get_port()
-
-rc = 1
-keepalive = 60
-connect_packet = mosq_test.gen_connect("test-helper", keepalive=keepalive)
-connack_packet = mosq_test.gen_connack(rc=0)
-
-mid = 312
-publish_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=2, mid=mid, payload="disconnect-message")
-pubrec_packet = mosq_test.gen_pubrec(mid)
-pubrel_packet = mosq_test.gen_pubrel(mid)
-pubcomp_packet = mosq_test.gen_pubcomp(mid)
-
-sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port, connack_error="helper connack")
-sock.send(publish_packet)
-
-if mosq_test.expect_packet(sock, "helper pubrec", pubrec_packet):
-    sock.send(pubrel_packet)
-
-    if mosq_test.expect_packet(sock, "helper pubcomp", pubcomp_packet):
-        rc = 0
-
-sock.close()
-
-exit(rc)
-

test/broker/06-bridge-br2b-disconnect-qos2.py

@@ -4,7 +4,7 @@
 
 from mosq_test_helper import *
 
-def write_config(filename, port1, port2):
+def write_config(filename, port1, port2, protocol_version):
     with open(filename, 'w') as f:
         f.write("port %d\n" % (port2))
         f.write("\n")
@@ -13,104 +13,131 @@ def write_config(filename, port1, port2):
         f.write("topic bridge/# both 2\n")
         f.write("notifications false\n")
         f.write("restart_timeout 5\n")
+        f.write("bridge_protocol_version %s\n" % (protocol_version))
 
 (port1, port2) = mosq_test.get_port(2)
 conf_file = os.path.basename(__file__).replace('.py', '.conf')
-write_config(conf_file, port1, port2)
 
-rc = 1
-keepalive = 60
-client_id = socket.gethostname()+".bridge_sample"
-connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=128+4)
-connack_packet = mosq_test.gen_connack(rc=0)
+def do_test(proto_ver):
+    if proto_ver == 4:
+        bridge_protocol = "mqttv311"
+        proto_ver_connect = 128+4
+    else:
+        bridge_protocol = "mqttv50"
+        proto_ver_connect = 5
+
+    write_config(conf_file, port1, port2, bridge_protocol)
+
+    rc = 1
+    keepalive = 60
+    client_id = socket.gethostname()+".bridge_sample"
+    connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=proto_ver_connect)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+
+    mid = 1
+    subscribe_packet = mosq_test.gen_subscribe(mid, "bridge/#", 2, proto_ver=proto_ver)
+    suback_packet = mosq_test.gen_suback(mid, 2, proto_ver=proto_ver)
+
+    mid = 3
+    subscribe2_packet = mosq_test.gen_subscribe(mid, "bridge/#", 2, proto_ver=proto_ver)
+    suback2_packet = mosq_test.gen_suback(mid, 2, proto_ver=proto_ver)
+
+    mid = 4
+    subscribe3_packet = mosq_test.gen_subscribe(mid, "bridge/#", 2, proto_ver=proto_ver)
+    suback3_packet = mosq_test.gen_suback(mid, 2, proto_ver=proto_ver)
+
+    mid = 2
+    publish_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=2, mid=mid, payload="disconnect-message", proto_ver=proto_ver)
+    publish_dup_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=2, mid=mid, payload="disconnect-message", dup=True, proto_ver=proto_ver)
+    pubrec_packet = mosq_test.gen_pubrec(mid, proto_ver=proto_ver)
+    pubrel_packet = mosq_test.gen_pubrel(mid, proto_ver=proto_ver)
+    pubcomp_packet = mosq_test.gen_pubcomp(mid, proto_ver=proto_ver)
+
+    ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    ssock.settimeout(40)
+    ssock.bind(('', port1))
+    ssock.listen(5)
+
+    broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
 
-mid = 1
-subscribe_packet = mosq_test.gen_subscribe(mid, "bridge/#", 2)
-suback_packet = mosq_test.gen_suback(mid, 2)
-
-mid = 3
-subscribe2_packet = mosq_test.gen_subscribe(mid, "bridge/#", 2)
-suback2_packet = mosq_test.gen_suback(mid, 2)
-
-mid = 4
-subscribe3_packet = mosq_test.gen_subscribe(mid, "bridge/#", 2)
-suback3_packet = mosq_test.gen_suback(mid, 2)
-
-mid = 2
-publish_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=2, mid=mid, payload="disconnect-message")
-publish_dup_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=2, mid=mid, payload="disconnect-message", dup=True)
-pubrec_packet = mosq_test.gen_pubrec(mid)
-pubrel_packet = mosq_test.gen_pubrel(mid)
-pubcomp_packet = mosq_test.gen_pubcomp(mid)
+    try:
+        (bridge, address) = ssock.accept()
+        bridge.settimeout(20)
 
-ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-ssock.settimeout(40)
-ssock.bind(('', port1))
-ssock.listen(5)
+        if mosq_test.expect_packet(bridge, "connect", connect_packet):
+            bridge.send(connack_packet)
 
-broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
+            if mosq_test.expect_packet(bridge, "subscribe", subscribe_packet):
+                bridge.send(suback_packet)
 
-try:
-    (bridge, address) = ssock.accept()
-    bridge.settimeout(20)
+                # Helper
+                helper_connect_packet = mosq_test.gen_connect("test-helper", keepalive=keepalive, proto_ver=proto_ver)
+                helper_connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
 
-    if mosq_test.expect_packet(bridge, "connect", connect_packet):
-        bridge.send(connack_packet)
+                mid = 312
+                helper_publish_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=2, mid=mid, payload="disconnect-message", proto_ver=proto_ver)
+                helper_pubrec_packet = mosq_test.gen_pubrec(mid=mid, proto_ver=proto_ver)
+                helper_pubrel_packet = mosq_test.gen_pubrel(mid=mid, proto_ver=proto_ver)
+                helper_pubcomp_packet = mosq_test.gen_pubcomp(mid=mid, proto_ver=proto_ver)
 
-        if mosq_test.expect_packet(bridge, "subscribe", subscribe_packet):
-            bridge.send(suback_packet)
+                helper_sock = mosq_test.do_client_connect(helper_connect_packet, helper_connack_packet, port=port2, connack_error="helper connack")
+                mosq_test.do_send_receive(helper_sock, helper_publish_packet, helper_pubrec_packet, "helper pubrec")
+                mosq_test.do_send_receive(helper_sock, helper_pubrel_packet, helper_pubcomp_packet, "helper pubcomp")
+                helper_sock.close()
+                # End helper
 
-            pub = subprocess.Popen(['./06-bridge-br2b-disconnect-qos2-helper.py', str(port2)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-            if pub.wait():
-                exit(1)
-            (stdo, stde) = pub.communicate()
 
-            if mosq_test.expect_packet(bridge, "publish", publish_packet):
-                bridge.close()
+                if mosq_test.expect_packet(bridge, "publish", publish_packet):
+                    bridge.close()
 
-                (bridge, address) = ssock.accept()
-                bridge.settimeout(20)
+                    (bridge, address) = ssock.accept()
+                    bridge.settimeout(20)
 
-                if mosq_test.expect_packet(bridge, "connect", connect_packet):
-                    bridge.send(connack_packet)
+                    if mosq_test.expect_packet(bridge, "connect", connect_packet):
+                        bridge.send(connack_packet)
 
-                    if mosq_test.expect_packet(bridge, "2nd subscribe", subscribe2_packet):
-                        bridge.send(suback2_packet)
+                        if mosq_test.expect_packet(bridge, "2nd subscribe", subscribe2_packet):
+                            bridge.send(suback2_packet)
 
-                        if mosq_test.expect_packet(bridge, "2nd publish", publish_dup_packet):
-                            bridge.send(pubrec_packet)
+                            if mosq_test.expect_packet(bridge, "2nd publish", publish_dup_packet):
+                                bridge.send(pubrec_packet)
 
-                            if mosq_test.expect_packet(bridge, "pubrel", pubrel_packet):
-                                bridge.close()
+                                if mosq_test.expect_packet(bridge, "pubrel", pubrel_packet):
+                                    bridge.close()
 
-                                (bridge, address) = ssock.accept()
-                                bridge.settimeout(20)
+                                    (bridge, address) = ssock.accept()
+                                    bridge.settimeout(20)
 
-                                if mosq_test.expect_packet(bridge, "connect", connect_packet):
-                                    bridge.send(connack_packet)
+                                    if mosq_test.expect_packet(bridge, "connect", connect_packet):
+                                        bridge.send(connack_packet)
 
-                                    if mosq_test.expect_packet(bridge, "3rd subscribe", subscribe3_packet):
-                                        bridge.send(suback3_packet)
+                                        if mosq_test.expect_packet(bridge, "3rd subscribe", subscribe3_packet):
+                                            bridge.send(suback3_packet)
 
-                                        if mosq_test.expect_packet(bridge, "2nd pubrel", pubrel_packet):
-                                            bridge.send(pubcomp_packet)
-                                            rc = 0
+                                            if mosq_test.expect_packet(bridge, "2nd pubrel", pubrel_packet):
+                                                bridge.send(pubcomp_packet)
+                                                rc = 0
 
-    bridge.close()
-finally:
-    os.remove(conf_file)
-    try:
         bridge.close()
-    except NameError:
-        pass
-
-    broker.terminate()
-    broker.wait()
-    (stdo, stde) = broker.communicate()
-    if rc:
-        print(stde.decode('utf-8'))
-    ssock.close()
-
-exit(rc)
+    finally:
+        os.remove(conf_file)
+        try:
+            bridge.close()
+        except NameError:
+            pass
+
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        ssock.close()
+        if rc:
+            print(stde.decode('utf-8'))
+            exit(rc)
+
+
+do_test(proto_ver=4)
+do_test(proto_ver=5)
+
+exit(0)
 

test/broker/06-bridge-br2b-remapping.py

@@ -4,7 +4,7 @@
 
 from mosq_test_helper import *
 
-def write_config(filename, port1, port2):
+def write_config(filename, port1, port2, protocol_version):
     with open(filename, 'w') as f:
         f.write("port %d\n" % (port2))
         f.write("\n")
@@ -19,30 +19,15 @@ def write_config(filename, port1, port2):
         f.write("topic +/value out 0 local5/top remote5/tip\n")
         f.write("notifications false\n")
         f.write("restart_timeout 5\n")
+        f.write("bridge_protocol_version %s\n" % (protocol_version))
 
 (port1, port2) = mosq_test.get_port(2)
 conf_file = os.path.basename(__file__).replace('.py', '.conf')
-write_config(conf_file, port1, port2)
 
-rc = 1
-keepalive = 60
-client_id = socket.gethostname()+".bridge_sample"
-connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=128+4)
-connack_packet = mosq_test.gen_connack(rc=0)
 
-client_connect_packet = mosq_test.gen_connect("pub-test", keepalive=keepalive)
-client_connack_packet = mosq_test.gen_connack(rc=0)
+def inner_test(bridge, sock, proto_ver):
+    global connect_packet, connack_packet
 
-ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-ssock.settimeout(4)
-ssock.bind(('', port1))
-ssock.listen(5)
-
-broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
-
-
-def test(bridge, sock):
     if not mosq_test.expect_packet(bridge, "connect", connect_packet):
         return 1
 
@@ -65,12 +50,12 @@ def test(bridge, sock):
     for (local_topic, remote_topic) in cases:
         mid += 1
         local_publish_packet = mosq_test.gen_publish(
-            local_topic, qos=0, mid=mid, payload=''
+            local_topic, qos=0, mid=mid, payload='', proto_ver=proto_ver
         )
         sock.send(local_publish_packet)
         if remote_topic:
             remote_publish_packet = mosq_test.gen_publish(
-                remote_topic, qos=0, mid=mid, payload=''
+                remote_topic, qos=0, mid=mid, payload='', proto_ver=proto_ver
             )
             match = mosq_test.expect_packet(bridge, "publish", remote_publish_packet)
             if not match:
@@ -86,31 +71,66 @@ def test(bridge, sock):
                 ))
     return 0
 
-try:
-    (bridge, address) = ssock.accept()
-    bridge.settimeout(2)
 
-    sock = mosq_test.do_client_connect(
-        client_connect_packet, client_connack_packet,
-        port=port2,
-    )
+def do_test(proto_ver):
+    global connect_packet, connack_packet
+
+    if proto_ver == 4:
+        bridge_protocol = "mqttv311"
+        proto_ver_connect = 128+4
+    else:
+        bridge_protocol = "mqttv50"
+        proto_ver_connect = 5
+
+    write_config(conf_file, port1, port2, bridge_protocol)
+
+    rc = 1
+    keepalive = 60
+    client_id = socket.gethostname()+".bridge_sample"
+    connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=proto_ver_connect)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
 
-    rc = test(bridge, sock)
+    client_connect_packet = mosq_test.gen_connect("pub-test", keepalive=keepalive, proto_ver=proto_ver)
+    client_connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+
+    ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    ssock.settimeout(4)
+    ssock.bind(('', port1))
+    ssock.listen(5)
+
+    broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
 
-    sock.close()
-    bridge.close()
-finally:
-    os.remove(conf_file)
     try:
-        bridge.close()
-    except NameError:
-        pass
+        (bridge, address) = ssock.accept()
+        bridge.settimeout(2)
 
-    broker.terminate()
-    broker.wait()
-    (stdo, stde) = broker.communicate()
-    if rc:
-        print(stde.decode('utf-8'))
-    ssock.close()
+        sock = mosq_test.do_client_connect(
+            client_connect_packet, client_connack_packet,
+            port=port2,
+        )
+
+        rc = inner_test(bridge, sock, proto_ver)
 
-exit(rc)
+        sock.close()
+        bridge.close()
+    finally:
+        os.remove(conf_file)
+        try:
+            bridge.close()
+        except NameError:
+            pass
+
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        ssock.close()
+        if rc:
+            print(stde.decode('utf-8'))
+            exit(rc)
+
+
+do_test(proto_ver=4)
+do_test(proto_ver=5)
+
+exit(0)

test/broker/06-bridge-fail-persist-resend-qos1.py

@@ -4,7 +4,7 @@
 
 from mosq_test_helper import *
 
-def write_config(filename, port1, port2):
+def write_config(filename, port1, port2, protocol_version):
     with open(filename, 'w') as f:
         f.write("port %d\n" % (port2))
         f.write("\n")
@@ -17,70 +17,86 @@ def write_config(filename, port1, port2):
         f.write("notifications false\n")
         f.write("restart_timeout 5\n")
         f.write("try_private false\n")
+        f.write("bridge_protocol_version %s\n" % (protocol_version))
+
 
 (port1, port2) = mosq_test.get_port(2)
 conf_file = os.path.basename(__file__).replace('.py', '.conf')
-write_config(conf_file, port1, port2)
 
-rc = 1
-keepalive = 60
-connect_packet = mosq_test.gen_connect("bridge-u-test", keepalive=keepalive)
-connack_packet = mosq_test.gen_connack(rc=0)
 
-mid = 180
-mid_unknown = 2000
+def do_test(proto_ver):
+    if proto_ver == 4:
+        bridge_protocol = "mqttv311"
+        proto_ver_connect = 128+4
+    else:
+        bridge_protocol = "mqttv50"
+        proto_ver_connect = 5
+
+    write_config(conf_file, port1, port2, bridge_protocol)
+
+    rc = 1
+    keepalive = 60
+    connect_packet = mosq_test.gen_connect("bridge-u-test", keepalive=keepalive, proto_ver=proto_ver)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+
+    mid = 180
+    mid_unknown = 2000
+
+    publish_packet = mosq_test.gen_publish("bridge/unknown/qos1", qos=1, payload="bridge-message", mid=mid, proto_ver=proto_ver)
+    puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
+    puback_packet_unknown = mosq_test.gen_puback(mid_unknown, proto_ver=proto_ver)
 
-publish_packet = mosq_test.gen_publish("bridge/unknown/qos1", qos=1, payload="bridge-message", mid=mid)
-puback_packet = mosq_test.gen_puback(mid)
-puback_packet_unknown = mosq_test.gen_puback(mid_unknown)
 
+    unsubscribe_packet = mosq_test.gen_unsubscribe(1, "bridge/#", proto_ver=proto_ver)
+    unsuback_packet = mosq_test.gen_unsuback(1, proto_ver=proto_ver)
 
-unsubscribe_packet = mosq_test.gen_unsubscribe(1, "bridge/#")
-unsuback_packet = mosq_test.gen_unsuback(1)
 
+    if os.environ.get('MOSQ_USE_VALGRIND') is not None:
+        sleep_time = 5
+    else:
+        sleep_time = 0.5
 
-if os.environ.get('MOSQ_USE_VALGRIND') is not None:
-    sleep_time = 5
-else:
-    sleep_time = 0.5
 
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    sock.settimeout(10)
+    sock.bind(('', port1))
+    sock.listen(5)
 
-sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-sock.settimeout(10)
-sock.bind(('', port1))
-sock.listen(5)
+    broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
+    time.sleep(sleep_time)
 
-broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
-time.sleep(sleep_time)
+    try:
+        (conn, address) = sock.accept()
+        conn.settimeout(20)
 
-try:
-    (conn, address) = sock.accept()
-    conn.settimeout(20)
+        if mosq_test.expect_packet(conn, "connect", connect_packet):
+            conn.send(connack_packet)
 
-    if mosq_test.expect_packet(conn, "connect", connect_packet):
-        conn.send(connack_packet)
+            if mosq_test.expect_packet(conn, "unsubscribe", unsubscribe_packet):
+                conn.send(unsuback_packet)
 
-        if mosq_test.expect_packet(conn, "unsubscribe", unsubscribe_packet):
-            conn.send(unsuback_packet)
+                # Send the unexpected puback packet
+                conn.send(puback_packet_unknown)
 
-            # Send the unexpected puback packet
-            conn.send(puback_packet_unknown)
+                # Send a legitimate publish packet to verify everything is still ok
+                conn.send(publish_packet)
 
-            # Send a legitimate publish packet to verify everything is still ok
-            conn.send(publish_packet)
+                if mosq_test.expect_packet(conn, "puback", puback_packet):
+                    rc = 0
 
-            if mosq_test.expect_packet(conn, "puback", puback_packet):
-                rc = 0
+    finally:
+        os.remove(conf_file)
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        sock.close()
+        if rc:
+            print(stde.decode('utf-8'))
+            exit(rc)
 
-finally:
-    os.remove(conf_file)
-    broker.terminate()
-    broker.wait()
-    (stdo, stde) = broker.communicate()
-    if rc:
-        print(stde.decode('utf-8'))
-    sock.close()
+do_test(proto_ver=4)
+do_test(proto_ver=5)
 
-exit(rc)
+exit(0)
 

test/broker/06-bridge-fail-persist-resend-qos2.py

@@ -4,7 +4,7 @@
 
 from mosq_test_helper import *
 
-def write_config(filename, port1, port2):
+def write_config(filename, port1, port2, protocol_version):
     with open(filename, 'w') as f:
         f.write("port %d\n" % (port2))
         f.write("\n")
@@ -17,83 +17,99 @@ def write_config(filename, port1, port2):
         f.write("notifications false\n")
         f.write("restart_timeout 5\n")
         f.write("try_private false\n")
+        f.write("bridge_protocol_version %s\n" % (protocol_version))
 
 (port1, port2) = mosq_test.get_port(2)
 conf_file = os.path.basename(__file__).replace('.py', '.conf')
-write_config(conf_file, port1, port2)
 
-rc = 1
-keepalive = 60
-connect_packet = mosq_test.gen_connect("bridge-u-test", keepalive=keepalive)
-connack_packet = mosq_test.gen_connack(rc=0)
 
-mid = 180
-mid_unknown = 2000
+def do_test(proto_ver):
+    if proto_ver == 4:
+        bridge_protocol = "mqttv311"
+        proto_ver_connect = 4
+    else:
+        bridge_protocol = "mqttv50"
+        proto_ver_connect = 5
 
-publish_packet = mosq_test.gen_publish("bridge/unknown/qos2", qos=1, payload="bridge-message", mid=mid)
-puback_packet = mosq_test.gen_puback(mid)
+    write_config(conf_file, port1, port2, bridge_protocol)
 
-pubrec_packet_unknown1 = mosq_test.gen_pubrec(mid_unknown+1)
-pubrel_packet_unknown1 = mosq_test.gen_pubrel(mid_unknown+1)
+    rc = 1
+    keepalive = 60
+    connect_packet = mosq_test.gen_connect("bridge-u-test", keepalive=keepalive, proto_ver=proto_ver_connect)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
 
-pubrel_packet_unknown2 = mosq_test.gen_pubrel(mid_unknown+2)
-pubcomp_packet_unknown2 = mosq_test.gen_pubcomp(mid_unknown+2)
+    mid = 180
+    mid_unknown = 2000
 
-pubcomp_packet_unknown3 = mosq_test.gen_pubcomp(mid_unknown+3)
+    publish_packet = mosq_test.gen_publish("bridge/unknown/qos2", qos=1, payload="bridge-message", mid=mid, proto_ver=proto_ver)
+    puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
 
+    pubrec_packet_unknown1 = mosq_test.gen_pubrec(mid_unknown+1, proto_ver=proto_ver)
+    pubrel_packet_unknown1 = mosq_test.gen_pubrel(mid_unknown+1, proto_ver=proto_ver)
 
-unsubscribe_packet = mosq_test.gen_unsubscribe(1, "bridge/#")
-unsuback_packet = mosq_test.gen_unsuback(1)
+    pubrel_packet_unknown2 = mosq_test.gen_pubrel(mid_unknown+2, proto_ver=proto_ver)
+    pubcomp_packet_unknown2 = mosq_test.gen_pubcomp(mid_unknown+2, proto_ver=proto_ver)
 
+    pubcomp_packet_unknown3 = mosq_test.gen_pubcomp(mid_unknown+3, proto_ver=proto_ver)
 
-if os.environ.get('MOSQ_USE_VALGRIND') is not None:
-    sleep_time = 5
-else:
-    sleep_time = 0.5
 
+    unsubscribe_packet = mosq_test.gen_unsubscribe(1, "bridge/#", proto_ver=proto_ver)
+    unsuback_packet = mosq_test.gen_unsuback(1, proto_ver=proto_ver)
 
-sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-sock.settimeout(10)
-sock.bind(('', port1))
-sock.listen(5)
 
-broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
-time.sleep(sleep_time)
+    if os.environ.get('MOSQ_USE_VALGRIND') is not None:
+        sleep_time = 5
+    else:
+        sleep_time = 0.5
 
-try:
-    (conn, address) = sock.accept()
-    conn.settimeout(20)
 
-    if mosq_test.expect_packet(conn, "connect", connect_packet):
-        conn.send(connack_packet)
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    sock.settimeout(10)
+    sock.bind(('', port1))
+    sock.listen(5)
 
-        if mosq_test.expect_packet(conn, "unsubscribe", unsubscribe_packet):
-            conn.send(unsuback_packet)
+    broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
+    time.sleep(sleep_time)
 
-            # Send the unexpected pubrec packet
-            conn.send(pubrec_packet_unknown1)
-            if mosq_test.expect_packet(conn, "pubrel", pubrel_packet_unknown1):
+    try:
+        (conn, address) = sock.accept()
+        conn.settimeout(20)
 
-                conn.send(pubrel_packet_unknown2)
-                if mosq_test.expect_packet(conn, "pubcomp", pubcomp_packet_unknown2):
+        if mosq_test.expect_packet(conn, "connect", connect_packet):
+            conn.send(connack_packet)
 
-                    conn.send(pubcomp_packet_unknown3)
+            if mosq_test.expect_packet(conn, "unsubscribe", unsubscribe_packet):
+                conn.send(unsuback_packet)
 
-                    # Send a legitimate publish packet to verify everything is still ok
-                    conn.send(publish_packet)
+                # Send the unexpected pubrec packet
+                conn.send(pubrec_packet_unknown1)
+                if mosq_test.expect_packet(conn, "pubrel", pubrel_packet_unknown1):
+
+                    conn.send(pubrel_packet_unknown2)
+                    if mosq_test.expect_packet(conn, "pubcomp", pubcomp_packet_unknown2):
+
+                        conn.send(pubcomp_packet_unknown3)
+
+                        # Send a legitimate publish packet to verify everything is still ok
+                        conn.send(publish_packet)
         
-                    if mosq_test.expect_packet(conn, "puback", puback_packet):
-                        rc = 0
-
-finally:
-    os.remove(conf_file)
-    broker.terminate()
-    broker.wait()
-    (stdo, stde) = broker.communicate()
-    if rc:
-        print(stde.decode('utf-8'))
-    sock.close()
-
-exit(rc)
+                        if mosq_test.expect_packet(conn, "puback", puback_packet):
+                            rc = 0
+
+    finally:
+        os.remove(conf_file)
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        sock.close()
+        if rc:
+            print(stde.decode('utf-8'))
+            exit(rc)
+
+
+do_test(proto_ver=4)
+do_test(proto_ver=5)
+
+exit(0)
 

test/broker/06-bridge-per-listener-settings.py

@@ -4,7 +4,7 @@
 
 from mosq_test_helper import *
 
-def write_config(filename, port1, port2, port3):
+def write_config(filename, port1, port2, port3, protocol_version):
     with open(filename, 'w') as f:
         f.write("per_listener_settings true\n")
         f.write("port %d\n" % (port2))
@@ -20,30 +20,16 @@ def write_config(filename, port1, port2, port3):
         f.write("topic clients/total in 0 test/mosquitto/org $SYS/broker/\n")
         f.write("notifications false\n")
         f.write("restart_timeout 5\n")
+        f.write("bridge_protocol_version %s\n" % (protocol_version))
 
 (port1, port2, port3) = mosq_test.get_port(3)
 conf_file = os.path.basename(__file__).replace('.py', '.conf')
-write_config(conf_file, port1, port2, port3)
 
-rc = 1
-keepalive = 60
-client_id = socket.gethostname()+".bridge_sample"
-connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=128+4)
-connack_packet = mosq_test.gen_connack(rc=0)
 
-client_connect_packet = mosq_test.gen_connect("pub-test", keepalive=keepalive)
-client_connack_packet = mosq_test.gen_connack(rc=0)
 
-ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-ssock.settimeout(4)
-ssock.bind(('', port1))
-ssock.listen(5)
+def inner_test(bridge, sock, proto_ver):
+    global connect_packet, connack_packet
 
-broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
-
-
-def test(bridge, sock):
     if not mosq_test.expect_packet(bridge, "connect", connect_packet):
         return 1
     bridge.send(connack_packet)
@@ -58,15 +44,15 @@ def test(bridge, sock):
     ]
     for pattern in ("remote/topic/#", "remote2/topic/prefix/#", "remote3/topic/+/value"):
         mid += 1
-        subscribe_packet = mosq_test.gen_subscribe(mid, pattern, 0)
-        suback_packet = mosq_test.gen_suback(mid, 0)
+        subscribe_packet = mosq_test.gen_subscribe(mid, pattern, 0, proto_ver=proto_ver)
+        suback_packet = mosq_test.gen_suback(mid, 0, proto_ver=proto_ver)
         if not mosq_test.expect_packet(bridge, "subscribe", subscribe_packet):
             return 1
         bridge.send(suback_packet)
 
     mid += 1
-    subscribe_packet = mosq_test.gen_subscribe(mid, "#", 0)
-    suback_packet = mosq_test.gen_suback(mid, 0)
+    subscribe_packet = mosq_test.gen_subscribe(mid, "#", 0, proto_ver=proto_ver)
+    suback_packet = mosq_test.gen_suback(mid, 0, proto_ver=proto_ver)
     sock.send(subscribe_packet)
     if not mosq_test.expect_packet(sock, "suback", suback_packet):
         return 1
@@ -86,11 +72,9 @@ def test(bridge, sock):
     for (local_topic, remote_topic) in cases:
         mid += 1
         remote_publish_packet = mosq_test.gen_publish(
-            remote_topic, qos=0, mid=mid, payload=''
-        )
+            remote_topic, qos=0, mid=mid, payload='', proto_ver=proto_ver)
         local_publish_packet = mosq_test.gen_publish(
-            local_topic, qos=0, mid=mid, payload=''
-        )
+            local_topic, qos=0, mid=mid, payload='', proto_ver=proto_ver)
 
         bridge.send(remote_publish_packet)
         match = mosq_test.expect_packet(sock, "publish", local_publish_packet)
@@ -101,31 +85,66 @@ def test(bridge, sock):
             return 1
     return 0
 
-try:
-    (bridge, address) = ssock.accept()
-    bridge.settimeout(2)
 
-    sock = mosq_test.do_client_connect(
-        client_connect_packet, client_connack_packet,
-        port=port2,
-    )
+def do_test(proto_ver):
+    global connect_packet, connack_packet
+
+    if proto_ver == 4:
+        bridge_protocol = "mqttv311"
+        proto_ver_connect = 128+4
+    else:
+        bridge_protocol = "mqttv50"
+        proto_ver_connect = 5
+
+    write_config(conf_file, port1, port2, port3, bridge_protocol)
+    rc = 1
+    keepalive = 60
+    client_id = socket.gethostname()+".bridge_sample"
+    connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=proto_ver_connect)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+
+    client_connect_packet = mosq_test.gen_connect("pub-test", keepalive=keepalive, proto_ver=proto_ver)
+    client_connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+
+    ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    ssock.settimeout(4)
+    ssock.bind(('', port1))
+    ssock.listen(5)
+
+    broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port2, use_conf=True)
 
-    rc = test(bridge, sock)
 
-    sock.close()
-    bridge.close()
-finally:
-    os.remove(conf_file)
     try:
-        bridge.close()
-    except NameError:
-        pass
+        (bridge, address) = ssock.accept()
+        bridge.settimeout(2)
 
-    broker.terminate()
-    broker.wait()
-    (stdo, stde) = broker.communicate()
-    if rc:
-        print(stde.decode('utf-8'))
-    ssock.close()
+        sock = mosq_test.do_client_connect(
+            client_connect_packet, client_connack_packet,
+            port=port2,
+        )
+
+        rc = inner_test(bridge, sock, proto_ver)
 
-exit(rc)
+        sock.close()
+        bridge.close()
+    finally:
+        os.remove(conf_file)
+        try:
+            bridge.close()
+        except NameError:
+            pass
+
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        ssock.close()
+        if rc:
+            print(stde.decode('utf-8'))
+            exit(rc)
+
+
+do_test(proto_ver=4)
+do_test(proto_ver=5)
+
+exit(0)

test/broker/06-bridge-reconnect-local-out-helper.py

@@ -1,26 +0,0 @@
-#!/usr/bin/env python3
-
-from mosq_test_helper import *
-
-port = mosq_test.get_port()
-
-rc = 1
-keepalive = 60
-connect_packet = mosq_test.gen_connect("test-helper", keepalive=keepalive)
-connack_packet = mosq_test.gen_connack(rc=0)
-
-publish_packet = mosq_test.gen_publish("bridge/reconnect", qos=1, mid=1, payload="bridge-reconnect-message")
-puback_packet = mosq_test.gen_puback(mid=1)
-
-disconnect_packet = mosq_test.gen_disconnect()
-
-sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port, connack_error="helper connack")
-mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback")
-
-sock.send(disconnect_packet)
-rc = 0
-
-sock.close()
-
-exit(rc)
-

test/broker/06-bridge-reconnect-local-out.py

@@ -5,7 +5,7 @@
 
 from mosq_test_helper import *
 
-def write_config(filename, port1, port2):
+def write_config(filename, port1, port2, protocol_version):
     with open(filename, 'w') as f:
         f.write("port %d\n" % (port2))
         f.write("\n")
@@ -15,82 +15,106 @@ def write_config(filename, port1, port2):
         f.write("connection bridge_sample\n")
         f.write("address 127.0.0.1:%d\n" % (port1))
         f.write("topic bridge/# out\n")
+        f.write("bridge_protocol_version %s\n" % (protocol_version))
 
 (port1, port2) = mosq_test.get_port(2)
 conf_file = '06-bridge-reconnect-local-out.conf'
-write_config(conf_file, port1, port2)
-
-rc = 1
-keepalive = 60
-connect_packet = mosq_test.gen_connect("bridge-reconnect-test", keepalive=keepalive)
-connack_packet = mosq_test.gen_connack(rc=0)
-
-mid = 180
-subscribe_packet = mosq_test.gen_subscribe(mid, "bridge/#", 0)
-suback_packet = mosq_test.gen_suback(mid, 0)
-publish_packet = mosq_test.gen_publish("bridge/reconnect", qos=0, payload="bridge-reconnect-message")
-
-try:
-    os.remove('mosquitto-%d.db' % (port1))
-except OSError:
-    pass
-
-broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port1, use_conf=False)
-
-local_cmd = ['../../src/mosquitto', '-c', '06-bridge-reconnect-local-out.conf']
-local_broker = mosq_test.start_broker(cmd=local_cmd, filename=os.path.basename(__file__)+'_local1', use_conf=False, port=port2)
-if os.environ.get('MOSQ_USE_VALGRIND') is not None:
-    time.sleep(5)
-else:
-    time.sleep(0.5)
-local_broker.terminate()
-local_broker.wait()
-if os.environ.get('MOSQ_USE_VALGRIND') is not None:
-    time.sleep(5)
-else:
-    time.sleep(0.5)
-local_broker = mosq_test.start_broker(cmd=local_cmd, filename=os.path.basename(__file__)+'_local2', port=port2)
-if os.environ.get('MOSQ_USE_VALGRIND') is not None:
-    time.sleep(5)
-else:
-    time.sleep(0.5)
-
-pub = None
-try:
-    sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port1)
-    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
-    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
-
-    pub = subprocess.Popen(['./06-bridge-reconnect-local-out-helper.py', str(port2)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    pub.wait()
-    (stdo, stde) = pub.communicate()
-    # Should have now received a publish command
-
-    if mosq_test.expect_packet(sock, "publish", publish_packet):
-        rc = 0
-
-    sock.close()
-finally:
-    os.remove(conf_file)
-    time.sleep(1)
-    broker.terminate()
-    broker.wait()
-    (stdo, stde) = broker.communicate()
-    if rc:
-        print(stde.decode('utf-8'))
-    local_broker.terminate()
-    local_broker.wait()
-    if rc:
-        (stdo, stde) = local_broker.communicate()
-        print(stde.decode('utf-8'))
-        if pub:
-            (stdo, stde) = pub.communicate()
-            print(stdo.decode('utf-8'))
+
+
+def do_test(proto_ver):
+    if proto_ver == 4:
+        bridge_protocol = "mqttv311"
+        proto_ver_connect = 128+4
+    else:
+        bridge_protocol = "mqttv50"
+        proto_ver_connect = 5
+
+    write_config(conf_file, port1, port2, bridge_protocol)
+
+    rc = 1
+    keepalive = 60
+    connect_packet = mosq_test.gen_connect("bridge-reconnect-test", keepalive=keepalive, proto_ver=proto_ver_connect)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+
+    mid = 180
+    subscribe_packet = mosq_test.gen_subscribe(mid, "bridge/#", 0, proto_ver=proto_ver)
+    suback_packet = mosq_test.gen_suback(mid, 0, proto_ver=proto_ver)
+    publish_packet = mosq_test.gen_publish("bridge/reconnect", qos=0, payload="bridge-reconnect-message", proto_ver=proto_ver)
 
     try:
         os.remove('mosquitto-%d.db' % (port1))
     except OSError:
         pass
 
-exit(rc)
+    broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port1, use_conf=False)
+
+    local_cmd = ['../../src/mosquitto', '-c', '06-bridge-reconnect-local-out.conf']
+    local_broker = mosq_test.start_broker(cmd=local_cmd, filename=os.path.basename(__file__)+'_local1', use_conf=False, port=port2)
+    if os.environ.get('MOSQ_USE_VALGRIND') is not None:
+        time.sleep(5)
+    else:
+        time.sleep(0.5)
+    local_broker.terminate()
+    local_broker.wait()
+    if os.environ.get('MOSQ_USE_VALGRIND') is not None:
+        time.sleep(5)
+    else:
+        time.sleep(0.5)
+    local_broker = mosq_test.start_broker(cmd=local_cmd, filename=os.path.basename(__file__)+'_local2', port=port2)
+    if os.environ.get('MOSQ_USE_VALGRIND') is not None:
+        time.sleep(5)
+    else:
+        time.sleep(0.5)
+
+    pub = None
+    try:
+        sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port1)
+        mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
+        mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
+
+        # Helper
+        helper_connect_packet = mosq_test.gen_connect("test-helper", keepalive=keepalive, proto_ver=proto_ver)
+        helper_connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+        helper_publish_packet = mosq_test.gen_publish("bridge/reconnect", qos=1, mid=1, payload="bridge-reconnect-message", proto_ver=proto_ver)
+        helper_puback_packet = mosq_test.gen_puback(mid=1, proto_ver=proto_ver)
+        helper_disconnect_packet = mosq_test.gen_disconnect(proto_ver=proto_ver)
+        helper_sock = mosq_test.do_client_connect(helper_connect_packet, helper_connack_packet, port=port2, connack_error="helper connack")
+        mosq_test.do_send_receive(helper_sock, helper_publish_packet, helper_puback_packet, "puback")
+        helper_sock.send(helper_disconnect_packet)
+        helper_sock.close()
+        # End of helper
+
+        # Should have now received a publish command
+        if mosq_test.expect_packet(sock, "publish", publish_packet):
+            rc = 0
+
+        sock.close()
+    finally:
+        os.remove(conf_file)
+        time.sleep(1)
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        if rc:
+            print(stde.decode('utf-8'))
+        local_broker.terminate()
+        local_broker.wait()
+        try:
+            os.remove('mosquitto-%d.db' % (port1))
+        except OSError:
+            pass
+
+        if rc:
+            (stdo, stde) = local_broker.communicate()
+            print(stde.decode('utf-8'))
+            if pub:
+                (stdo, stde) = pub.communicate()
+                print(stdo.decode('utf-8'))
+            exit(rc)
+
+
+do_test(proto_ver=4)
+do_test(proto_ver=5)
+
+exit(0)