Merge pull request #395 from martindurant/source_refactor

refactor and make less tornado

Author: martindurant

.coveragerc

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

setup.cfg

@@ -4,7 +4,7 @@
 # https://flake8.readthedocs.io/en/latest/user/error-codes.html
 
 # Note: there cannot be spaces after comma's here
-exclude = __init__.py, compatibility.py
+exclude = __init__.py,tests
 ignore =
     # Extra space in brackets
     E20,

streamz/compatibility.py

@@ -23,7 +23,7 @@
 
 from collections.abc import Iterable
 
-from .compatibility import get_thread_identity
+from threading import get_ident as get_thread_identity
 from .orderedweakset import OrderedWeakrefSet
 
 no_default = '--no-default--'
@@ -512,8 +512,10 @@ def disconnect(self, downstream):
 
     @property
     def upstream(self):
-        if len(self.upstreams) != 1:
+        if len(self.upstreams) > 1:
             raise ValueError("Stream has multiple upstreams")
+        elif len(self.upstreams) == 0:
+            return None
         else:
             return self.upstreams[0]
 
@@ -535,6 +537,13 @@ 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/core.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)