refactor PeriodicCallback but keep alias for now

Author: Martin Durant

streamz/sinks.py

@@ -107,8 +107,5 @@ def __init__(self, upstream, file, end="\n", mode="a", **kwargs):
         weakref.finalize(self, self._fp.close)
         super().__init__(upstream, **kwargs)
 
-    def __del__(self):
-        self._fp.close()
-
     def update(self, x, who=None, metadata=None):
         self._fp.write(x + self._end)

streamz/sources.py

@@ -2,25 +2,11 @@
 from glob import glob
 import os
 import time
-import tornado.ioloop
 from tornado import gen
 
 from .core import Stream, convert_interval, RefCounter
 
 
-def PeriodicCallback(callback, callback_time, asynchronous=False, **kwargs):
-    source = Stream(asynchronous=asynchronous)
-
-    def _():
-        result = callback()
-        source._emit(result)
-
-    pc = tornado.ioloop.PeriodicCallback(_, callback_time, **kwargs)
-    pc.start()
-    pc.start()
-    return source
-
-
 def sink_to_file(filename, upstream, mode='w', prefix='', suffix='\n', flush=False):
     file = open(filename, mode=mode)
 
@@ -34,6 +20,17 @@ def write(text):
 
 
 class Source(Stream):
+    """Start node for a set of Streams
+
+    Source nodes emit data into other nodes. They typically get this data
+    by polling external sources, and are necessarily run by an event loop.
+
+    Parameters
+    ----------
+    start: bool
+        Whether to call the run method immediately. If False, nothing
+        will happen until ``source.start()`` is called.
+    """
     _graphviz_shape = 'doubleoctagon'
 
     def __init__(self, start=False, **kwargs):
@@ -49,15 +46,68 @@ def stop(self):
             self.stopped = True
 
     def start(self):
-        """start polling"""
-        self.stopped = False
-        self.started = True
-        self.loop.add_callback(self.run)
+        """start polling
+
+        If already running, this has no effect. If the source was started and then
+        stopped again, this will restart the ``self.run`` coroutine.
+        """
+        if self.stopped:
+            self.stopped = False
+            self.started = True
+            self.loop.add_callback(self.run)
 
     async def run(self):
+        """This coroutine will be invoked by start() and emit all data
+
+        You might either overrive ``_run()`` when all logic can be contained
+        there, or override this method directly.
+
+        Note the use of ``.stopped`` to halt the coroutine, whether or not
+
+        """
         while not self.stopped:
             await self._run()
 
+    async def _run(self):
+        """This is the functionality to run on each cycle
+
+        Typically this may be used for polling some external IO source
+        or time-based data emission. You might choose to include an
+        ``await asyncio.sleep()`` for the latter.
+        """
+        raise NotImplementedError
+
+
+@Stream.register_api(staticmethod)
+class from_periodic(Source):
+    """Generate data from a function on given period
+
+    cf ``streamz.dataframe.PeriodicDataFrame``
+
+    Parameters
+    ----------
+    callback: callable
+        Function to call on each iteration. Takes no arguments.
+    poll_interval: float
+        Time to sleep between calls (s)
+    """
+
+    def __init__(self, callback, poll_interval=0.1, **kwargs):
+        self._cb = callback
+        self._poll = poll_interval
+        super().__init__(**kwargs)
+
+    async def _run(self):
+        await asyncio.gather(self._emit(self._cb()))
+        await asyncio.sleep(self._poll)
+
+
+def PeriodicCallback(callback, callback_time, asynchronous=False, **kwargs):  # pragma: no cover
+    """For backward compatibility - please use Stream.from_periodic"""
+    if kwargs:
+        callback = lambda: callback(**kwargs)
+    return Stream.from_periodic(callback, callback_time, asynchronous=asynchronous)
+
 
 @Stream.register_api(staticmethod)
 class from_textfile(Source):

streamz/tests/test_core.py

@@ -5,19 +5,18 @@
 import operator
 from operator import add
 import os
-from time import time, sleep
+from time import sleep
 import sys
 
 import pytest
 
-from tornado import gen
 from tornado.queues import Queue
 from tornado.ioloop import IOLoop
 
 import streamz as sz
 
-from streamz import Stream, RefCounter
-from streamz.sources import sink_to_file, PeriodicCallback
+from streamz import RefCounter
+from streamz.sources import sink_to_file
 from streamz.utils_test import (inc, double, gen_test, tmpfile, captured_logger,   # noqa: F401
         clean, await_for, metadata, wait_for)  # noqa: F401
 from distributed.utils_test import loop   # noqa: F401
@@ -421,7 +420,7 @@ def test_timed_window_ref_counts():
 @gen_test()
 def test_timed_window_metadata():
     source = Stream()
-    L = metadata(source.timed_window(0.01)).sink_to_list()
+    L = metadata(source.timed_window(0.06)).sink_to_list()
 
     source.emit(0)
     source.emit(1, metadata=[{'v': 1}])
@@ -482,7 +481,8 @@ def test_sink_to_file():
 @gen_test()
 def test_counter():
     counter = itertools.count()
-    source = PeriodicCallback(lambda: next(counter), 0.001, asynchronous=True)
+    source = Stream.from_periodic(lambda: next(counter), 0.001, asynchronous=True,
+                                  start=True)
     L = source.sink_to_list()
     yield gen.sleep(0.05)