fix stop()

Author: Martin Durant

streamz/core.py

@@ -560,13 +560,6 @@ def remove(self, predicate):
         """ Only pass through elements for which the predicate returns False """
         return self.filter(lambda x: not predicate(x))
 
-    def stop(self):
-        """Call on any stream node to halt all upstream sources"""
-        prev, s = self.upstream, self
-        while s:
-            prev, s = s, s.upstream
-        prev.stopped = True
-
     @property
     def scan(self):
         return self.accumulate

streamz/dataframe/core.py

@@ -876,7 +876,7 @@ def __init__(self, datafn=random_datablock, interval='500ms', dask=False,
         self.loop = source.loop
         self.interval = pd.Timedelta(interval).total_seconds()
         self.source = source
-        self.continue_ = [True]
+        self.continue_ = [False]  # like the oppose of self.stopped
         self.kwargs = kwargs
 
         stream = self.source.map(
@@ -889,8 +889,10 @@ def __init__(self, datafn=random_datablock, interval='500ms', dask=False,
             self.start()
 
     def start(self):
-        self.loop.add_callback(self._cb, self.interval, self.source,
-                               self.continue_)
+        if not self.continue_[0]:
+            self.continue_[0] = True
+            self.loop.add_callback(self._cb, self.interval, self.source,
+                                   self.continue_)
 
     def __del__(self):
         self.stop()
@@ -924,8 +926,9 @@ class Random(PeriodicDataFrame):
     """
 
     def __init__(self, freq='100ms', interval='500ms', dask=False,
-                 datafn=random_datablock):
-        super(Random, self).__init__(datafn, interval, dask, freq=pd.Timedelta(freq))
+                 start=True, datafn=random_datablock):
+        super(Random, self).__init__(datafn, interval, dask, start,
+                                     freq=pd.Timedelta(freq))
 
 
 _stream_types['streaming'].append((is_dataframe_like, DataFrame))

streamz/dataframe/tests/test_dataframes.py

@@ -557,7 +557,6 @@ def test_gc():
     a = DataFrame({'volatility': sdf.x.rolling('100ms').var(),
                             'sub': sdf.x - sdf.x.rolling('100ms').mean()})
     n = len(sdf.stream.downstreams)
-    yield gen.sleep(0.1)
     a = DataFrame({'volatility': sdf.x.rolling('100ms').var(),
                             'sub': sdf.x - sdf.x.rolling('100ms').mean()})
     yield gen.sleep(0.1)
@@ -566,6 +565,7 @@ def test_gc():
     yield gen.sleep(0.1)
     a = DataFrame({'volatility': sdf.x.rolling('100ms').var(),
                             'sub': sdf.x - sdf.x.rolling('100ms').mean()})
+    yield gen.sleep(0.1)
 
     assert len(sdf.stream.downstreams) == n
     del a