.buffer: make up_to optional

Author: ebonnal

streamable/_afunctions.py

@@ -33,7 +33,7 @@
 
 def buffer(
     aiterator: AsyncIterator[T],
-    up_to: int,
+    up_to: Optional[int] = None,
 ) -> AsyncIterator[T]:
     return _aiterators.BufferAsyncIterator(aiterator, up_to)
 

streamable/_aiterators.py

@@ -37,7 +37,7 @@
     FIFOFutureResults,
     FutureResults,
 )
-from streamable._tools._async import AsyncFunction, empty_aiter
+from streamable._tools._async import AsyncFunction, NoopSemaphore, empty_aiter
 from streamable._tools._context import noop_context_manager
 from streamable._tools._error import ExceptionContainer, RaisingAsyncIterator
 
@@ -59,7 +59,7 @@ class _BufferAsyncIterable(AsyncIterable[Union[T, ExceptionContainer]]):
     def __init__(
         self,
         iterator: AsyncIterator[T],
-        up_to: int,
+        up_to: Optional[int],
     ) -> None:
         self.iterator = iterator
         self.up_to = up_to
@@ -76,7 +76,9 @@ def _lazy_buffer(self) -> "asyncio.Queue[Union[T, ExceptionContainer]]":
     @property
     def _lazy_slots(self) -> asyncio.Semaphore:
         if not self._slots:
-            self._slots = asyncio.Semaphore(self.up_to)
+            self._slots = (
+                asyncio.Semaphore(self.up_to) if self.up_to else NoopSemaphore()
+            )
         return self._slots
 
     async def _buffer_upstream(self) -> None:
@@ -115,7 +117,7 @@ class BufferAsyncIterator(RaisingAsyncIterator[T]):
     def __init__(
         self,
         iterator: AsyncIterator[T],
-        up_to: int,
+        up_to: Optional[int],
     ) -> None:
         super().__init__(_BufferAsyncIterable(iterator, up_to).__aiter__())
 

streamable/_functions.py

@@ -30,7 +30,7 @@
 
 def buffer(
     iterator: Iterator[T],
-    up_to: int,
+    up_to: Optional[int] = None,
 ) -> Iterator[T]:
     return _iterators.BufferIterator(iterator, up_to)
 

streamable/_iterators.py

@@ -27,6 +27,7 @@
 
 from streamable._tools._context import noop_context_manager
 from streamable._tools._observation import Observation
+from streamable._tools._threading import NoopSemaphore
 from streamable._tools._sentinel import STOP_ITERATION
 from streamable._tools._validation import validate_sync_flatten_iterable
 
@@ -55,11 +56,11 @@ class _BufferIterable(Iterable[Union[T, ExceptionContainer]]):
     def __init__(
         self,
         iterator: Iterator[T],
-        up_to: int,
+        up_to: Optional[int],
     ) -> None:
         self.iterator = iterator
         self._buffer: "queue.Queue[Union[T, ExceptionContainer]]" = queue.Queue()
-        self._slots = Semaphore(up_to)
+        self._slots = Semaphore(up_to) if up_to else NoopSemaphore()
         self._stopped = False
 
     def _buffer_upstream(self) -> None:
@@ -99,7 +100,7 @@ class BufferIterator(RaisingIterator[T]):
     def __init__(
         self,
         iterator: Iterator[T],
-        up_to: int,
+        up_to: Optional[int],
     ) -> None:
         super().__init__(_BufferIterable(iterator, up_to).__iter__())
 

streamable/_stream.py

@@ -300,17 +300,15 @@ def cast(self, into: Type[U]) -> "stream[U]":
 
     def buffer(
         self,
-        up_to: int,
+        up_to: Optional[int] = None,
     ) -> "stream[T]":
         """
-        Buffer upstream elements into a bounded queue (max size ``up_to``), via a background task.
-
-        Allow to decouple the upstream production rate from the downstream consumption rate.
+        Buffer upstream elements into a queue, via a background task, decoupling upstream production rate from downstream consumption rate.
 
         The background task is a thread during a sync iteration, and an async task during an async iteration.
 
         Args:
-            up_to (``int``): The buffer size. Must be >= 1. When reached, upstream pulling pauses until an element is yielded out of the buffer.
+            up_to (``int | None``): The buffer size, must be >= 1 when set. When reached, upstream pulling pauses until an element is yielded out of the buffer.
 
         Returns:
             ``stream[T]``: Upstream with buffering.
@@ -327,7 +325,8 @@ def buffer(
             time.sleep(1e-3)
             assert pulled == [0, 1, 2, 3, 4, 5]
         """
-        validate_int(up_to, gte=1, name="up_to")
+        if up_to is not None:
+            validate_int(up_to, gte=1, name="up_to")
         return BufferStream(self, up_to)
 
     @overload
@@ -1082,7 +1081,7 @@ class BufferStream(DownStream[T, T]):
     def __init__(
         self,
         upstream: stream[T],
-        up_to: int,
+        up_to: Optional[int],
     ) -> None:
         super().__init__(upstream)
         self._up_to = up_to

streamable/_tools/_async.py

@@ -1,9 +1,11 @@
+from asyncio import Semaphore
 from typing import (
     Any,
     AsyncIterator,
     Awaitable,
     Callable,
     Coroutine,
+    Literal,
     TypeVar,
 )
 
@@ -25,3 +27,19 @@ async def awaitable_to_coroutine(aw: Awaitable[T]) -> T:
 async def empty_aiter() -> AsyncIterator[Any]:
     return
     yield  # pragma: no cover
+
+
+class NoopSemaphore(Semaphore):
+    __slots__ = ()
+
+    def __init__(self) -> None:
+        pass
+
+    async def acquire(self) -> Literal[True]:
+        return True
+
+    def release(self) -> None:
+        return
+
+    def locked(self) -> bool:
+        return False

tests/test_buffer.py

@@ -1,3 +1,4 @@
+import sys
 from typing import Any, AsyncIterable, Callable, Iterable, List
 
 import pytest
@@ -26,7 +27,7 @@ def test_buffer_preserves_elements(itype: IterableType) -> None:
     assert alist_or_list(ints.buffer(5), itype) == list(INTEGERS)
 
 
-@pytest.mark.parametrize("buffer_size", [1, 10])
+@pytest.mark.parametrize("buffer_size", [1, 10, None])
 @pytest.mark.parametrize(
     "itype, slow_identity",
     [(Iterable, slow_identity), (AsyncIterable, async_slow_identity)],
@@ -39,9 +40,13 @@ def test_buffer_size_is_respected(
     buffering_ints_iter = aiter_or_iter(buffering_ints, itype)
     assert buffered == []
     assert anext_or_next(buffering_ints_iter, itype) == 0
-    assert buffered == list(INTEGERS)[: buffer_size + 1]
+    assert (
+        buffered == list(INTEGERS)[: (buffer_size + 1) if buffer_size else sys.maxsize]
+    )
     assert anext_or_next(buffering_ints_iter, itype) == 1
-    assert buffered == list(INTEGERS)[: buffer_size + 2]
+    assert (
+        buffered == list(INTEGERS)[: (buffer_size + 2) if buffer_size else sys.maxsize]
+    )
 
 
 @pytest.mark.parametrize(

tests/test_tools.py

@@ -2,6 +2,7 @@
 import pytest
 from streamable._tools._func import sidify
 from streamable._tools._logging import logfmt_str_escape
+from streamable._tools import _async, _threading
 
 from typing import Any, Callable, List
 
@@ -88,3 +89,18 @@ def test_logfmt_str_escape():
     assert logfmt_str_escape("in ts") == '"in ts"'
     assert logfmt_str_escape("in\\ts") == r'"in\\ts"'
     assert logfmt_str_escape('"ints"') == r'"\"ints\""'
+
+
+def test_noop_semaphore() -> None:
+    semaphore = _threading.NoopSemaphore()
+    assert semaphore.acquire()
+    semaphore.release()
+    assert not semaphore.locked()
+
+
+@pytest.mark.asyncio
+async def test_noop_semaphore_async() -> None:
+    semaphore = _async.NoopSemaphore()
+    assert await semaphore.acquire()
+    semaphore.release()
+    assert not semaphore.locked()