From 4028ec9cb55ce7268baaba03030b4fc20a05816b Mon Sep 17 00:00:00 2001
From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com>
Date: Tue, 20 Oct 2026 07:33:29 +0000
Subject: [PATCH] BufferReader.read: timeout for blocking reads (#8)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
---
 stream2py/buffer_reader.py                     | 12 +++++++++++-
 stream2py/tests/test_buffer_reader_blocking.py | 14 ++++++++++++++
 2 files changed, 25 insertions(+), 1 deletion(-)

diff --git a/stream2py/buffer_reader.py b/stream2py/buffer_reader.py
index 369fe62..a3890f1 100644
--- a/stream2py/buffer_reader.py
+++ b/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:
diff --git a/stream2py/tests/test_buffer_reader_blocking.py b/stream2py/tests/test_buffer_reader_blocking.py
index f572994..16a7241 100644
--- a/stream2py/tests/test_buffer_reader_blocking.py
+++ b/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