X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=juju%2Futils.py;h=f4db66e723c850770b24ccfb90ce17484f72a517;hb=c8a68279c2043d7b3d3789a881826d52074af04e;hp=78322e7923f22f5f0afc8b5cdd26dafc324141c3;hpb=efca4234627865400e0d2ab74a0f77bd6988c397;p=osm%2FN2VC.git

diff --git a/juju/utils.py b/juju/utils.py
index 78322e7..f4db66e 100644
--- a/juju/utils.py
+++ b/juju/utils.py
@@ -1,10 +1,11 @@
 import asyncio
-import concurrent.futures
 import os
+from collections import defaultdict
+from functools import partial
 from pathlib import Path
 
 
-async def execute_process(*cmd, log=None):
+async def execute_process(*cmd, log=None, loop=None):
     '''
     Wrapper around asyncio.create_subprocess_exec.
 
@@ -14,7 +15,7 @@ async def execute_process(*cmd, log=None):
             stdin=asyncio.subprocess.PIPE,
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,
-            )
+            loop=loop)
     stdout, stderr = await p.communicate()
     if log:
         log.debug("Exec %s -> %d", cmd, p.returncode)
@@ -45,8 +46,26 @@ async def read_ssh_key(loop):
     can be passed on to a model.
 
     '''
-    ssh_key = await loop.run_in_executor(
-        concurrent.futures.ThreadPoolExecutor(),
-        _read_ssh_key
-    )
-    return ssh_key
+    return await loop.run_in_executor(None, _read_ssh_key)
+
+
+class IdQueue:
+    """
+    Wrapper around asyncio.Queue that maintains a separate queue for each ID.
+    """
+    def __init__(self, maxsize=0, *, loop=None):
+        self._queues = defaultdict(partial(asyncio.Queue, maxsize, loop=loop))
+
+    async def get(self, id):
+        value = await self._queues[id].get()
+        del self._queues[id]
+        if isinstance(value, Exception):
+            raise value
+        return value
+
+    async def put(self, id, value):
+        await self._queues[id].put(value)
+
+    async def put_all(self, value):
+        for queue in self._queues.values():
+            await queue.put(value)