await_for instead of sleep in test_from_file_end

Author: Martin Durant

streamz/sources.py

@@ -76,11 +76,13 @@ def __init__(self, f, poll_interval=0.100, delimiter='\n', start=False,
         self.poll_interval = poll_interval
         super(from_textfile, self).__init__(ensure_io_loop=True, **kwargs)
         self.stopped = True
+        self.started = False
         if start:
             self.start()
 
     def start(self):
         self.stopped = False
+        self.started = False
         self.loop.add_callback(self.do_poll)
 
     @gen.coroutine
@@ -89,7 +91,8 @@ def do_poll(self):
         if self.from_end:
             # this only happens when we are ready to read
             self.file.seek(0, 2)
-        while True:
+        while not self.stopped:
+            self.started = True
             line = self.file.read()
             if line:
                 buffer = buffer + line
@@ -100,8 +103,6 @@ def do_poll(self):
                         yield self._emit(part + self.delimiter)
             else:
                 yield gen.sleep(self.poll_interval)
-            if self.stopped:
-                break
 
 
 @Stream.register_api(staticmethod)

streamz/tests/test_core.py

@@ -799,7 +799,7 @@ def test_from_file_end():
             out = source.sink_to_list()
             source.start()
             assert out == []
-            yield gen.sleep(0.01)
+            yield await_for(lambda: source.started, 2, period=0.02)
 
             f.write('data2\n')
             f.flush()