Added test for out of order receive and fix bug creating task

[osm/N2VC.git] / juju / client / connection.py

diff --git a/juju/client/connection.py b/juju/client/connection.py
index b508a1a..82ec4fa 100644 (file)
--- a/juju/client/connection.py
+++ b/juju/client/connection.py
@@ -16,6 +16,7 @@ import yaml
 
 from juju import tag
 from juju.errors import JujuError, JujuAPIError, JujuConnectionError
+from juju.utils import IdQueue
 
 log = logging.getLogger("websocket")
 
@@ -52,6 +53,7 @@ class Connection:
         self.addr = None
         self.ws = None
         self.facades = {}
+        self.messages = IdQueue(loop=self.loop)
 
     @property
     def is_open(self):
@@ -74,17 +76,22 @@ class Connection:
         kw['loop'] = self.loop
         self.addr = url
         self.ws = await websockets.connect(url, **kw)
+        self.loop.create_task(self.receiver())
         log.info("Driver connected to juju %s", url)
         return self
 
     async def close(self):
         await self.ws.close()
 
-    async def recv(self):
-        result = await self.ws.recv()
-        if result is not None:
-            result = json.loads(result)
-        return result
+    async def recv(self, request_id):
+        return await self.messages.get(request_id)
+
+    async def receiver(self):
+        while self.is_open:
+            result = await self.ws.recv()
+            if result is not None:
+                result = json.loads(result)
+                await self.messages.put(result['request-id'], result)
 
     async def rpc(self, msg, encoder=None):
         self.__request_id__ += 1
@@ -95,7 +102,7 @@ class Connection:
             msg['version'] = self.facades[msg['type']]
         outgoing = json.dumps(msg, indent=2, cls=encoder)
         await self.ws.send(outgoing)
-        result = await self.recv()
+        result = await self.recv(msg['request-id'])
 
         if not result:
             return result