default downsteams is empty list, override destroy for sinks

Author: roveo

streamz/core.py

@@ -165,8 +165,10 @@ def __init__(self, upstream=None, upstreams=None, stream_name=None,
         self.downstreams = OrderedWeakrefSet()
         if upstreams is not None:
             self.upstreams = list(upstreams)
-        else:
+        elif upstream is not None:
             self.upstreams = [upstream]
+        else:
+            self.upstreams = []
 
         self._set_asynchronous(asynchronous)
         self._set_loop(loop)
@@ -236,10 +238,7 @@ def _inform_asynchronous(self, asynchronous):
     def _add_upstream(self, upstream):
         """Add upstream to current upstreams, this method is overridden for
         classes which handle stream specific buffers/caches"""
-        if self.upstreams == [None]:
-            self.upstreams[0] = upstream
-        else:
-            self.upstreams.append(upstream)
+        self.upstreams.append(upstream)
 
     def _add_downstream(self, downstream):
         """Add downstream to current downstreams"""
@@ -252,10 +251,7 @@ def _remove_downstream(self, downstream):
     def _remove_upstream(self, upstream):
         """Remove upstream from current upstreams, this method is overridden for
         classes which handle stream specific buffers/caches"""
-        if len(self.upstreams) == 1:
-            self.upstreams[0] = [None]
-        else:
-            self.upstreams.remove(upstream)
+        self.upstreams.remove(upstream)
 
     @classmethod
     def register_api(cls, modifier=identity, attribute_name=None):
@@ -527,8 +523,8 @@ def destroy(self, streams=None):
         if streams is None:
             streams = self.upstreams
         for upstream in list(streams):
-            upstream.downstreams.remove(self)
-            self.upstreams.remove(upstream)
+            upstream._remove_downstream(self)
+            self._remove_upstream(upstream)
 
     def scatter(self, **kwargs):
         from .dask import scatter

streamz/sinks.py

@@ -67,6 +67,10 @@ def update(self, x, who=None, metadata=None):
         else:
             return []
 
+    def destroy(self):
+        super().destroy()
+        _global_sinks.remove(self)
+
 
 @Stream.register_api()
 class sink_to_textfile(Sink):

streamz/tests/test_core.py

@@ -1080,7 +1080,7 @@ def test_connect():
     # connect assumes this default behaviour
     # of stream initialization
     assert not source_downstream.downstreams
-    assert source_downstream.upstreams == [None]
+    assert source_downstream.upstreams == []
 
     # initialize the second stream to connect to
     source_upstream = Stream()

streamz/tests/test_sinks.py

@@ -1,3 +1,5 @@
+import weakref
+
 import pytest
 from streamz import Stream
 from streamz.sinks import _global_sinks
@@ -56,3 +58,13 @@ def test_sink_to_textfile_closes():
 
         with pytest.raises(ValueError, match=r"I/O operation on closed file\."):
             fp.write(".")
+
+
+def test_sink_destroy():
+    source = Stream()
+    sink = source.sink(lambda x: None)
+    ref = weakref.ref(sink)
+    sink.destroy()
+    del sink
+
+    assert ref() is None