_ConcurrentMapAsyncIterable: run_in_executor

Author: ebonnal

streamable/aiterators.py

@@ -1,7 +1,6 @@
 import asyncio
 import datetime
 import multiprocessing
-import queue
 import time
 from abc import ABC, abstractmethod
 from collections import defaultdict, deque
@@ -39,9 +38,7 @@
 from streamable.util.constants import NO_REPLACEMENT
 from streamable.util.futuretools import (
     FDFOAsyncFutureResultCollection,
-    FDFOOSFutureResultCollection,
     FIFOAsyncFutureResultCollection,
-    FIFOOSFutureResultCollection,
     FutureResult,
     FutureResultCollection,
 )
@@ -601,11 +598,13 @@ def _context_manager(self) -> ContextManager:
     @abstractmethod
     def _launch_task(self, elem: T) -> "Future[Union[U, ExceptionContainer]]": ...
 
-    # factory method
-    @abstractmethod
     def _future_result_collection(
         self,
-    ) -> FutureResultCollection[Union[U, ExceptionContainer]]: ...
+    ) -> FutureResultCollection[Union[U, ExceptionContainer]]:
+        if self.ordered:
+            return FIFOAsyncFutureResultCollection(asyncio.get_running_loop())
+        else:
+            return FDFOAsyncFutureResultCollection(asyncio.get_running_loop())
 
     async def _next_future(
         self,
@@ -676,17 +675,11 @@ def _safe_transformation(
             return ExceptionContainer(e)
 
     def _launch_task(self, elem: T) -> "Future[Union[U, ExceptionContainer]]":
-        return self.executor.submit(
-            self._safe_transformation, self.transformation, elem
-        )
-
-    def _future_result_collection(
-        self,
-    ) -> FutureResultCollection[Union[U, ExceptionContainer]]:
-        if self.ordered:
-            return FIFOOSFutureResultCollection()
-        return FDFOOSFutureResultCollection(
-            multiprocessing.Queue if self.via == "process" else queue.Queue
+        return cast(
+            "Future[Union[U, ExceptionContainer]]",
+            asyncio.get_running_loop().run_in_executor(
+                self.executor, self._safe_transformation, self.transformation, elem
+            ),
         )
 
 
@@ -745,14 +738,6 @@ def _launch_task(self, elem: T) -> "Future[Union[U, ExceptionContainer]]":
             asyncio.get_running_loop().create_task(self._safe_transformation(elem)),
         )
 
-    def _future_result_collection(
-        self,
-    ) -> FutureResultCollection[Union[U, ExceptionContainer]]:
-        if self.ordered:
-            return FIFOAsyncFutureResultCollection(asyncio.get_running_loop())
-        else:
-            return FDFOAsyncFutureResultCollection(asyncio.get_running_loop())
-
 
 class ConcurrentAMapAsyncIterator(_RaisingAsyncIterator[U]):
     def __init__(

tests/test_util.py

@@ -1,6 +1,8 @@
+import asyncio
 import unittest
 
 from streamable.util.functiontools import sidify, star
+from streamable.util.futuretools import FIFOOSFutureResultCollection, FutureResult
 
 
 class TestUtil(unittest.TestCase):
@@ -32,3 +34,12 @@ def mul(a: int, b: int) -> int:
             list(map(mul, enumerate(range(10)))),
             list(map(lambda x: x**2, range(10))),
         )
+
+    def test_os_future_result_collection_anext(self):
+        result = object()
+        future_results = FIFOOSFutureResultCollection()
+        future_results.add_future(FutureResult(result))
+        self.assertEqual(
+            asyncio.run(future_results.__anext__()),
+            result,
+        )