WIP idea about merging the calls

Author: dwskoog

streamz/core.py

@@ -1,4 +1,6 @@
 import asyncio
+import inspect
+import sys
 from collections import deque, defaultdict
 from datetime import timedelta
 from itertools import chain
@@ -705,8 +707,30 @@ def __init__(self, upstream, func, *args, **kwargs):
         stream_name = kwargs.pop('stream_name', None)
         self.kwargs = kwargs
         self.args = args
+        if inspect.iscoroutinefunction(func) or gen.is_coroutine_function(func):
+            self.work_queue = asyncio.Queue()
+        else:
+            self.work_queue = None
 
         Stream.__init__(self, upstream, stream_name=stream_name)
+        if self.work_queue:
+            self.cb_task = self.loop.asyncio_loop.create_task(self.cb())
+        else:
+            self.cb_task = None
+
+    async def cb(self):
+        while True:
+            if not self.work_queue:
+                assert self.work_queue, f"Async mapping engine should not have been started on {self}"
+                await asyncio.sleep(sys.float_info.max)  # Sleep forever in case assertion are turned off
+            coro, metadata = await self.work_queue.get()
+            try:
+                result = await coro
+            except Exception as e:
+                logger.exception(e)
+                raise
+            else:
+                return self._emit(result, metadata=metadata)
 
     def update(self, x, who=None, metadata=None):
         try:
@@ -715,7 +739,10 @@ def update(self, x, who=None, metadata=None):
             logger.exception(e)
             raise
         else:
-            return self._emit(result, metadata=metadata)
+            if self.work_queue:
+                return self.work_queue.put_nowait((result, metadata))
+            else:
+                return self._emit(result, metadata=metadata)
 
 
 @Stream.register_api()

streamz/tests/test_core.py

@@ -153,7 +153,7 @@ async def add_native(x=0, y=0):
         return x + y
 
     source = Stream(asynchronous=True)
-    L = source.map_async(add_tor, y=1).map_async(add_native, y=2).sink_to_list()
+    L = source.map(add_tor, y=1).map_async(add_native, y=2).sink_to_list()
 
     await source.emit(0)