coverage

Author: Martin Durant

.coveragerc

@@ -13,3 +13,4 @@ omit =
 exclude_lines =
     if __name__ == '__main__':
     pragma: no cover
+    NotImplementedError

streamz/sources.py

@@ -98,7 +98,7 @@ def __init__(self, callback, poll_interval=0.1, **kwargs):
         super().__init__(**kwargs)
 
     async def _run(self):
-        await asyncio.gather(self._emit(self._cb()))
+        await asyncio.gather(*self._emit(self._cb()))
         await asyncio.sleep(self._poll)
 
 
@@ -355,26 +355,30 @@ def __init__(self, cmd, open_kwargs=None, with_stderr=False, with_end=True,
         super().__init__(**kwargs)
 
     async def run(self):
+        import shlex
         import subprocess
         stderr = subprocess.STDOUT if self.with_stderr else None
         if isinstance(self.cmd, (list, tuple)):
             cmd, *args = self.cmd
         else:
-            cmd, args = self.cmd, ()
+            cmd, *args = shlex.split(self.cmd)
         process = await asyncio.create_subprocess_exec(
             cmd, *args, stdout=subprocess.PIPE,
             stderr=stderr, **self.open_kwargs)
         while not self.stopped:
             try:
                 out = await process.stdout.readuntil(b'\n')
             except asyncio.IncompleteReadError as err:
-                if self.with_end:
+                if self.with_end and err.partial:
                     out = err.partial
                 else:
                     break
+            if process.returncode is not None:
+                self.stopped = True
             await asyncio.gather(*self._emit(out))
-        process.terminate()
-        await process.wait()
+        if process.returncode is not None:
+            process.terminate()
+            await process.wait()
 
 
 @Stream.register_api(staticmethod)

streamz/tests/test_sources.py

@@ -1,10 +1,23 @@
+import asyncio
+import sys
+
 from flaky import flaky
 import pytest
 from streamz import Source
 from streamz.utils_test import wait_for, await_for, gen_test
 import socket
 
 
+def test_periodic():
+    s = Source.from_periodic(lambda: True)
+    l = s.sink_to_list()
+    assert s.stopped
+    s.start()
+    wait_for(lambda: l, 0.11, period=0.01)
+    wait_for(lambda: len(l) > 1, 0.11, period=0.01)
+    assert all(l)
+
+
 @flaky(max_runs=3, min_passes=1)
 def test_tcp():
     port = 9876
@@ -88,9 +101,23 @@ def test_http():
 
 @gen_test(timeout=60)
 def test_process():
-    import sys
-    import asyncio
-    cmd = ["python", "-c", "for i in range(4): print(i)"]
+    cmd = ["python", "-c", "for i in range(4): print(i, end='')"]
+    s = Source.from_process(cmd, with_end=True)
+    if sys.platform != "win32":
+        # don't know why - something with pytest and new processes
+        policy = asyncio.get_event_loop_policy()
+        watcher = asyncio.SafeChildWatcher()
+        policy.set_child_watcher(watcher)
+        watcher.attach_loop(s.loop.asyncio_loop)
+    out = s.sink_to_list()
+    s.start()
+    yield await_for(lambda: out == [b'0123'], timeout=5)
+    s.stop()
+
+
+@gen_test(timeout=60)
+def test_process_str():
+    cmd = 'python -c "for i in range(4): print(i)"'
     s = Source.from_process(cmd)
     if sys.platform != "win32":
         # don't know why - something with pytest and new processes