Create new map_async api

I tried map with a coroutine and it failed spectacularly:
```
    @gen_test()
    def test_map_async_tornado():
        @gen.coroutine
        def add_tor(x=0, y=0):
            return x + y

        source = Stream(asynchronous=True)
        L = source.map(add_tor, y=1).map(add_tor, y=2).sink_to_list()

        yield source.emit(0)

        yield gen.moment  # yield to the event loop to ensure it finished
>       assert L == [3]
E       assert [<Future finished exception=TypeError("unsupported operand type(s) for +: '_asyncio.Future' and 'int'")>] == [3]
E
E         At index 0 diff: <Future finished exception=TypeError("unsupported operand type(s) for +: '_asyncio.Future' and 'int'")> != 3
E
E         Full diff:
E           [
E         -     3,
E         +     <Future finished exception=TypeError("unsupported operand type(s) for +: '_asyncio.Future' and 'int'")>,
E           ]
```
So I made a new `map_async` that uses native asyncio plumbing to await
the coroutine before feeding it downstream.

Author: dwskoog

streamz/core.py

@@ -718,6 +718,61 @@ def update(self, x, who=None, metadata=None):
             return self._emit(result, metadata=metadata)
 
 
+@Stream.register_api()
+class map_async(Stream):
+    """ Apply an async function to every element in the stream
+
+    Parameters
+    ----------
+    func: async callable
+    *args :
+        The arguments to pass to the function.
+    **kwargs:
+        Keyword arguments to pass to func
+
+    Examples
+    --------
+    >>> async def mult(x, factor=1):
+    ...     return factor*x
+    >>> async def run():
+    ...     source = Stream(asynchronous=True)
+    ...     source.map_async(mult, factor=2).sink(print)
+    ...     for i in range(5):
+    ...         await source.emit(i)
+    >>> asyncio.run(run())
+    0
+    2
+    4
+    6
+    8
+
+    """
+    def __init__(self, upstream, func, *args, **kwargs):
+        self.func = func
+        stream_name = kwargs.pop('stream_name', None)
+        self.kwargs = kwargs
+        self.args = args
+        self.work_queue = asyncio.Queue()
+
+        Stream.__init__(self, upstream, stream_name=stream_name, ensure_io_loop=True)
+        self.cb_task = self.loop.asyncio_loop.create_task(self.cb())
+
+    def update(self, x, who=None, metadata=None):
+        coro = self.func(x, *self.args, **self.kwargs)
+        return self.work_queue.put_nowait((coro, metadata))
+
+    async def cb(self):
+        while True:
+            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)
+
+
 @Stream.register_api()
 class starmap(Stream):
     """ Apply a function to every element in the stream, splayed out

streamz/tests/test_core.py

@@ -126,6 +126,42 @@ def add(x=0, y=0):
     assert L[0] == 11
 
 
+@gen_test()
+def test_map_async_tornado():
+    @gen.coroutine
+    def add_tor(x=0, y=0):
+        return x + y
+
+    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()
+
+    yield source.emit(0)
+
+    yield gen.moment  # Must yield to the event loop to ensure it finished
+    assert L == [3]
+
+
+@pytest.mark.asyncio
+async def test_map_async():
+    @gen.coroutine
+    def add_tor(x=0, y=0):
+        return x + y
+
+    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()
+
+    await source.emit(0)
+
+    await asyncio.sleep(0)  # Must yield to the event loop to ensure it finished
+    assert L == [3]
+
+
 def test_map_args():
     source = Stream()
     L = source.map(operator.add, 10).sink_to_list()