BufferReader.read: timeout for blocking reads (#8)

stream2py/buffer_reader.py

@@ -389,6 +389,7 @@ def read(
         ignore_no_item_found=None,
         strict_n=None,
         blocking=False,
+        timeout=None,
     ):
         """Finds an item with a key greater than the last returned item.
         Raise ValueError if no item found with key above last item.
@@ -402,6 +403,9 @@ def read(
             instead of raising exception
         :param strict_n: if True, raise ValueError if n items are not available
         :param blocking: if True, wait to return next item if not yet available
+        :param timeout: only used when ``blocking`` is True. Maximum number of seconds
+            to wait; ``None`` (default) waits until an item is available or the stream
+            stops. When it runs out, returns None (as when the stream stops).
         :return: next item or list of next items if n > 1
         """
         n, ignore_no_item_found, strict_n = defaulted_values(
@@ -421,6 +425,7 @@ def read(
         # strict_n = strict_n or self._read_kwargs['strict_n']
 
         if blocking is True:
+            deadline = None if timeout is None else time.monotonic() + timeout
             while True:
                 _read = self.read(
                     n,
@@ -433,8 +438,13 @@ def read(
                     return _read
                 elif self.is_stopped:
                     return None
-                else:
+                elif deadline is None:
                     time.sleep(self._sleep_time_on_iter_none_s)
+                else:
+                    remaining = deadline - time.monotonic()
+                    if remaining <= 0:
+                        return None
+                    time.sleep(min(self._sleep_time_on_iter_none_s, remaining))
 
         with self._buffer.reader_lock() as reader:
             try:

stream2py/tests/test_buffer_reader_blocking.py

@@ -74,3 +74,17 @@ def blocking_read_after_stop():
     # Should have gotten None because buffer stopped
     assert len(results) == 1
     assert results[0] is None, "Blocking read should return None when stopped"
+
+
+def test_buffer_reader_blocking_timeout():
+    """A blocking read gives up after ``timeout`` seconds, and only then."""
+    with StreamBuffer(SimpleSourceReader(range(3)), maxlen=100) as buffer:
+        reader = buffer.mk_reader()
+        assert wait_until(lambda: len(buffer.source_buffer) >= 3)
+        # data available: timeout is irrelevant
+        assert reader.read(n=3, blocking=True, timeout=5) is not None
+        # no more data: returns None once the timeout has elapsed
+        start = time.monotonic()
+        assert reader.read(blocking=True, timeout=0.3) is None
+        elapsed = time.monotonic() - start
+        assert 0.3 <= elapsed < 2