Add docs

Author: Martin Durant

docs/source/api.rst

@@ -43,7 +43,10 @@ Stream
 .. automethod:: Stream.emit
 .. automethod:: Stream.frequencies
 .. automethod:: Stream.register_api
+.. automethod:: Stream.sink
 .. automethod:: Stream.sink_to_list
+.. automethod:: Stream.sink_to_textfile
+.. automethod:: Stream.to_websocket
 .. automethod:: Stream.update
 .. automethod:: Stream.visualize
 
@@ -56,6 +59,7 @@ Sources
    from_kafka
    from_kafka_batched
    from_process
+   from_websocket
    from_textfile
    from_tcp
    from_http_server

streamz/core.py

@@ -1,3 +1,4 @@
+import asyncio
 from collections import deque, defaultdict
 from datetime import timedelta
 import functools
@@ -485,15 +486,14 @@ def emit(self, x, asynchronous=False, metadata=None):
             finally:
                 thread_state.asynchronous = ts_async
         else:
-            @gen.coroutine
-            def _():
+            async def _():
                 thread_state.asynchronous = True
                 try:
-                    result = yield self._emit(x, metadata=metadata)
+                    result = await asyncio.gather(*self._emit(x, metadata=metadata))
                 finally:
                     del thread_state.asynchronous
+                return result
 
-                raise gen.Return(result)
             sync(self.loop, _)
 
     def update(self, x, who=None, metadata=None):

streamz/sinks.py

@@ -113,7 +113,22 @@ def update(self, x, who=None, metadata=None):
 
 
 @Stream.register_api()
-class sink_to_websocket(Sink):
+class to_websocket(Sink):
+    """Write bytes data to websocket
+
+    The websocket will be opened on first call, and kept open. Should
+    it close at some point, future writes will fail.
+
+    Requires the ``websockets`` package.
+
+    :param uri: str
+        Something like "ws://host:port". Use "wss:" to allow TLS.
+    :param ws_kwargs: dict
+        Further kwargs to pass to ``websockets.connect``, please
+        read its documentation.
+    :param kwargs:
+        Passed to superclass
+    """
 
     def __init__(self, upstream, uri, ws_kwargs=None, **kwargs):
         self.uri = uri
@@ -129,7 +144,7 @@ def update(self, x, who=None, metadata=None):
         yield self.ws.send(x)
 
     def destroy(self):
-        super(sink_to_websocket, self).destroy()
+        super().destroy()
         if self.ws is not None:
             sync(self.loop, self.ws.protocol.close)
         self.ws = None

streamz/sources.py

@@ -783,6 +783,23 @@ async def run(self):
 
 @Stream.register_api()
 class from_websocket(Source):
+    """Read binary data from a websocket
+
+    This source will accept connections on a given port and handle messages
+    coming in.
+
+    The websockets library must be installed.
+
+    :param host: str
+        Typically "localhost"
+    :param port: int
+        Which port to listen on (must be available)
+    :param serve_kwargs: dict
+        Passed to ``websockets.serve``
+    :param kwargs:
+        Passed to superclass
+    """
+
     def __init__(self, host, port, serve_kwargs=None, **kwargs):
         self.host = host
         self.port = port

streamz/tests/test_sinks.py

@@ -80,7 +80,7 @@ def test_ws_roundtrip():
 
     data = [b'0123'] * 4
     s = Stream.from_iterable(data)
-    s.sink_to_websocket("ws://localhost:8989")
+    s.to_websocket("ws://localhost:8989")
     s.start()
 
     wait_for(lambda: data == l, timeout=1)