add test: sink_to_textfile closes file when deleted

update docs

Author: roveo

docs/source/api.rst

@@ -26,6 +26,7 @@ Stream
    rate_limit
    scatter
    sink
+   sink_to_textfile
    slice
    sliding_window
    starmap
@@ -84,6 +85,7 @@ Definitions
 .. autofunction:: partition
 .. autofunction:: rate_limit
 .. autofunction:: sink
+.. autofunction:: sink_to_textfile
 .. autofunction:: sliding_window
 .. autofunction:: Stream
 .. autofunction:: timed_window

docs/source/plugins.rst

@@ -60,7 +60,7 @@ Different kinds of add-ons go into different entry point groups:
 =========== ======================= =====================
  Source      ``streamz.Source``      ``streamz.sources``
  Node        ``streamz.Stream``      ``streamz.nodes``
- Sink        ``streamz.Stream``      ``streamz.sinks``
+ Sink        ``streamz.Sink``        ``streamz.sinks``
 =========== ======================= =====================
 
 

streamz/sinks.py

@@ -55,15 +55,16 @@ def update(self, x, who=None, metadata=None):
 
 @Stream.register_api()
 class sink_to_textfile(Sink):
-    """ Write elements to a plain text file, one element per line. Type of elements
-        must be ``str``.
+    """ Write elements to a plain text file, one element per line.
+
+        Type of elements must be ``str``.
 
         Arguments
         ---------
         file: str or file-like
             File to write the elements to. ``str`` is treated as a file name to open.
-            If file-like, descriptor must be open in text mode.
-            Note that the file descriptor will be closed when sink is destroyed.
+            If file-like, descriptor must be open in text mode. Note that the file
+            descriptor will be closed when this sink is destroyed.
         end: str, optional
             This value will be written to the file after each element.
             Defaults to newline character.
@@ -74,7 +75,7 @@ class sink_to_textfile(Sink):
         Examples
         --------
         >>> source = Stream()
-        >>> source.map(str).sink_to_file("test.txt")
+        >>> source.map(str).sink_to_textfile("test.txt")
         >>> source.emit(0)
         >>> source.emit(1)
         >>> print(open("test.txt", "r").read())

streamz/tests/test_sinks.py

@@ -1,6 +1,8 @@
 from time import sleep
 
+import pytest
 from streamz import Stream
+from streamz.sinks import _global_sinks
 from streamz.utils_test import tmpfile
 
 
@@ -43,3 +45,15 @@ def test_sink_to_textfile_named():
         sleep(0.01)
 
         assert open(filename, "r").read() == "0\n1\n"
+
+
+def test_sink_to_textfile_closes():
+    source = Stream()
+    with tmpfile() as filename:
+        sink = source.sink_to_textfile(filename)
+        fp = sink._fp
+        _global_sinks.remove(sink)
+        del sink
+
+        with pytest.raises(ValueError):  # I/O operation on closed file
+            fp.write(".")