Fix issues #111 and #104;

gmqtt/__init__.py

@@ -13,7 +13,7 @@
     "Mikhail Turchunovich",
     "Elena Nikolaichik"
 ]
-__version__ = "0.6.5"
+__version__ = "0.6.6"
 
 
 __all__ = [

gmqtt/mqtt/connection.py

@@ -89,6 +89,7 @@ async def close(self):
         if self._keep_connection_callback:
             self._keep_connection_callback.cancel()
         self._transport.close()
+        await self._protocol.closed
 
     def is_closing(self):
         return self._transport.is_closing()

gmqtt/mqtt/protocol.py

@@ -16,7 +16,7 @@ def __init__(self, buffer_size=2**16, loop=None):
         self._connection = None
         self._transport = None
 
-        self._connected = asyncio.Event(loop=loop)
+        self._connected = asyncio.Event()
 
         reader = asyncio.StreamReader(limit=buffer_size, loop=loop)
         # this is bad hack for python 3.8
@@ -27,6 +27,10 @@ def __init__(self, buffer_size=2**16, loop=None):
     def set_connection(self, conn):
         self._connection = conn
 
+    @property
+    def closed(self):
+        return self._closed
+
     def _parse_packet(self):
         raise NotImplementedError
 
@@ -198,6 +202,7 @@ async def _read_loop(self):
     def connection_lost(self, exc):
         super(MQTTProtocol, self).connection_lost(exc)
         self._connection.put_package((MQTTCommands.DISCONNECT, b''))
+
         if self._read_loop_future is not None:
             self._read_loop_future.cancel()
             self._read_loop_future = None

tests/test_mqtt5.py

@@ -173,7 +173,7 @@ async def test_shared_subscriptions(init_clients):
     aclient.subscribe(shared_sub_topic)
     aclient.subscribe(TOPICS[0])
 
-    await   bclient.connect(host=host, port=port)
+    await bclient.connect(host=host, port=port)
     bclient.subscribe(shared_sub_topic)
     bclient.subscribe(TOPICS[0])
 
@@ -208,6 +208,8 @@ async def test_shared_subscriptions(init_clients):
     assert len(callback.messages) > 0
     assert len(callback2.messages) > 0
 
+    await pubclient.disconnect()
+
 
 @pytest.mark.asyncio
 async def test_assigned_clientid():