add breaking condition, Sink.destroy test uses the base Sink class

Author: roveo

streamz/sources.py

@@ -763,4 +763,7 @@ def start(self):
     @gen.coroutine
     def _run(self):
         for x in self._iterable:
+            if self.stopped:
+                break
             yield self._emit(x)
+        self.stopped = True

streamz/tests/test_sinks.py

@@ -2,7 +2,7 @@
 
 import pytest
 from streamz import Stream
-from streamz.sinks import _global_sinks
+from streamz.sinks import _global_sinks, Sink
 from streamz.utils_test import tmpfile
 
 
@@ -62,7 +62,7 @@ def test_sink_to_textfile_closes():
 
 def test_sink_destroy():
     source = Stream()
-    sink = source.sink(lambda x: None)
+    sink = Sink(source)
     ref = weakref.ref(sink)
     sink.destroy()
     del sink