Add test for asyncio mode.

examples/loop_asyncio.py

@@ -4,6 +4,7 @@ import asyncio
 import socket
 import uuid
 
+import context  # Ensures paho is in PYTHONPATH
 import paho.mqtt.client as mqtt
 
 client_id = 'paho-mqtt-python/issue72/' + str(uuid.uuid4())

test/lib/01-asyncio.py

+#!/usr/bin/env python3
+
+# Test whether asyncio works
+
+import context
+import paho_test
+
+rc = 1
+keepalive = 60
+connect_packet = paho_test.gen_connect("asyncio-test", keepalive=keepalive)
+connack_packet = paho_test.gen_connack(rc=0)
+
+subscribe_packet = paho_test.gen_subscribe(mid=1, topic=u"sub-test", qos=1)
+suback_packet = paho_test.gen_suback(mid=1, qos=1)
+
+unsubscribe_packet = paho_test.gen_unsubscribe(mid=2, topic=u"unsub-test")
+unsuback_packet = paho_test.gen_unsuback(mid=2)
+
+publish_packet = paho_test.gen_publish(u"b2c", qos=0, payload="msg".encode('utf-8'))
+
+publish_packet_in = paho_test.gen_publish(u"asyncio", qos=1, mid=3, payload="message".encode('utf-8'))
+puback_packet_in = paho_test.gen_puback(mid=3)
+
+disconnect_packet = paho_test.gen_disconnect()
+
+sock = paho_test.create_server_socket()
+
+client = context.start_client()
+
+try:
+    (conn, address) = sock.accept()
+    conn.settimeout(10)
+
+    if paho_test.expect_packet(conn, "connect", connect_packet):
+        conn.send(connack_packet)
+
+        if paho_test.expect_packet(conn, "subscribe", subscribe_packet):
+            conn.send(suback_packet)
+
+            if paho_test.expect_packet(conn, "unsubscribe", unsubscribe_packet):
+                conn.send(unsuback_packet)
+                conn.send(publish_packet)
+
+                if paho_test.expect_packet(conn, "publish", publish_packet_in):
+                    conn.send(puback_packet_in)
+
+                    if paho_test.expect_packet(conn, "disconnect", disconnect_packet):
+                        rc = 0
+
+    conn.close()
+finally:
+    client.terminate()
+    client.wait()
+    sock.close()
+
+exit(rc)

test/lib/Makefile

@@ -6,6 +6,7 @@ PYTHON?=python3
 all :
 
 test :
+	$(PYTHON) ./01-asyncio.py python/01-asyncio.test
 	$(PYTHON) ./01-decorators.py python/01-decorators.test
 	$(PYTHON) ./01-keepalive-pingreq.py python/01-keepalive-pingreq.test
 	$(PYTHON) ./01-no-clean-session.py python/01-no-clean-session.test

test/lib/python/01-asyncio.test

+#!/usr/bin/env python3
+
+import asyncio
+import socket
+import uuid
+
+import context  # Ensures paho is in PYTHONPATH
+import paho.mqtt.client as mqtt
+
+client_id = 'asyncio-test'
+
+class AsyncioHelper:
+    def __init__(self, loop, client):
+        self.loop = loop
+        self.client = client
+        self.client.on_socket_open = self.on_socket_open
+        self.client.on_socket_close = self.on_socket_close
+        self.client.on_socket_register_write = self.on_socket_register_write
+        self.client.on_socket_unregister_write = self.on_socket_unregister_write
+
+    def on_socket_open(self, client, userdata, sock):
+        def cb():
+            client.loop_read()
+
+        self.loop.add_reader(sock, cb)
+        self.misc = self.loop.create_task(self.misc_loop())
+
+    def on_socket_close(self, client, userdata, sock):
+        self.loop.remove_reader(sock)
+        self.misc.cancel()
+
+    def on_socket_register_write(self, client, userdata, sock):
+        def cb():
+            client.loop_write()
+
+        self.loop.add_writer(sock, cb)
+
+    def on_socket_unregister_write(self, client, userdata, sock):
+        self.loop.remove_writer(sock)
+
+    async def misc_loop(self):
+        while self.client.loop_misc() == mqtt.MQTT_ERR_SUCCESS:
+            try:
+                await asyncio.sleep(1)
+            except asyncio.CancelledError:
+                break
+
+
+class AsyncMqttExample:
+    def __init__(self, loop):
+        self.loop = loop
+        self.payload = ""
+        self.complete = False
+
+    def on_connect(self, client, obj, flags, rc):
+        client.subscribe("sub-test", 1)
+
+    def on_subscribe(self, client, obj, mid, granted_qos):
+        client.unsubscribe("unsub-test")
+
+    def on_unsubscribe(self, client, obj, mid):
+        self.payload = "message"
+
+    def on_message(self, client, obj, msg):
+        client.publish("asyncio", qos=1, payload=self.payload)
+
+    def on_publish(self, client, obj, mid):
+        client.disconnect()
+
+    def on_disconnect(self, client, userdata, rc):
+        self.disconnected.set_result(rc)
+
+    async def main(self):
+        global rc
+        self.disconnected = self.loop.create_future()
+
+        self.client = mqtt.Client(client_id=client_id)
+        self.client.on_connect = self.on_connect
+        self.client.on_message = self.on_message
+        self.client.on_publish = self.on_publish
+        self.client.on_subscribe = self.on_subscribe
+        self.client.on_unsubscribe = self.on_unsubscribe
+        self.client.on_disconnect = self.on_disconnect
+
+        aioh = AsyncioHelper(self.loop, self.client)
+
+        self.client.connect('localhost', 1888, 60)
+        self.client.socket().setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 2048)
+
+        await self.disconnected
+        rc = 0
+
+rc = 1
+loop = asyncio.get_event_loop()
+loop.run_until_complete(AsyncMqttExample(loop).main())
+loop.close()
+exit(rc)