Add until operator

README.rst

@@ -61,7 +61,7 @@ The `stream operators`_ are separated in 7 categories:
 +--------------------+---------------------------------------------------------------------------------------+
 | **transformation** | map_, enumerate_, starmap_, cycle_, chunks_                                           |
 +--------------------+---------------------------------------------------------------------------------------+
-| **selection**      | take_, takelast_, skip_, skiplast_, getitem_, filter_, takewhile_, dropwhile_         |
+| **selection**      | take_, takelast_, skip_, skiplast_, getitem_, filter_, until_, takewhile_, dropwhile_ |
 +--------------------+---------------------------------------------------------------------------------------+
 | **combination**    | map_, zip_, merge_, chain_, ziplatest_                                                |
 +--------------------+---------------------------------------------------------------------------------------+
@@ -165,8 +165,9 @@ Vincent Michel: vxgmichel@gmail.com
 .. _skiplast: http://aiostream.readthedocs.io/en/latest/operators.html#aiostream.stream.skiplast
 .. _getitem: http://aiostream.readthedocs.io/en/latest/operators.html#aiostream.stream.getitem
 .. _filter: http://aiostream.readthedocs.io/en/latest/operators.html#aiostream.stream.filter
-.. _dropwhile: http://aiostream.readthedocs.io/en/latest/operators.html#aiostream.stream.dropwhile
+.. _until: http://aiostream.readthedocs.io/en/latest/operators.html#aiostream.stream.until
 .. _takewhile: http://aiostream.readthedocs.io/en/latest/operators.html#aiostream.stream.takewhile
+.. _dropwhile: http://aiostream.readthedocs.io/en/latest/operators.html#aiostream.stream.dropwhile
 
 .. _chain: http://aiostream.readthedocs.io/en/latest/operators.html#aiostream.stream.chain
 .. _zip: http://aiostream.readthedocs.io/en/latest/operators.html#aiostream.stream.zip

aiostream/stream/select.py

@@ -9,7 +9,7 @@
 from ..core import operator, streamcontext
 
 __all__ = ['take', 'takelast', 'skip', 'skiplast',
-           'getitem', 'filter', 'dropwhile', 'takewhile']
+           'getitem', 'filter', 'until', 'dropwhile', 'takewhile']
 
 
 @operator(pipable=True)
@@ -190,10 +190,35 @@ async def filter(source, func):
                 yield item
 
 
+@operator(pipable=True)
+async def until(source, func):
+    """Forward an asynchronous sequence until a condition is met.
+
+    Contrary to the ``takewhile`` operator, the last tested element is included
+    in the sequence.
+
+    The given function takes the item as an argument and returns a boolean
+    corresponding to the condition to meet. The function can either be
+    synchronous or asynchronous.
+    """
+    iscorofunc = asyncio.iscoroutinefunction(func)
+    async with streamcontext(source) as streamer:
+        async for item in streamer:
+            result = func(item)
+            if iscorofunc:
+                result = await result
+            yield item
+            if result:
+                return
+
+
 @operator(pipable=True)
 async def takewhile(source, func):
     """Forward an asynchronous sequence while a condition is met.
 
+    Contrary to the ``until`` operator, the last tested element is not included
+    in the sequence.
+
     The given function takes the item as an argument and returns a boolean
     corresponding to the condition to meet. The function can either be
     synchronous or asynchronous.

docs/operators.rst

@@ -93,6 +93,8 @@ Selection operators
 
 .. autoclass:: filter
 
+.. autoclass:: until
+
 .. autoclass:: takewhile
 
 .. autoclass:: dropwhile

docs/table.rst.inc

@@ -3,7 +3,7 @@
 +--------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------+
 | **transformation** | :class:`map`, :class:`enumerate`, :class:`starmap`, :class:`cycle`, :class:`chunks`                                                                                |
 +--------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------+
-| **selection**      | :class:`take`, :class:`takelast`, :class:`skip`, :class:`skiplast`, :class:`getitem`, :class:`filter`, :class:`takewhile`, :class:`dropwhile`                      |
+| **selection**      | :class:`take`, :class:`takelast`, :class:`skip`, :class:`skiplast`, :class:`getitem`, :class:`filter`, :class:`until`, :class:`takewhile`, :class:`dropwhile`      |
 +--------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------+
 | **combination**    | :class:`map`, :class:`zip`, :class:`merge`, :class:`chain`, :class:`ziplatest`                                                                                     |
 +--------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------+

tests/test_select.py

@@ -162,6 +162,26 @@ async def afunc(x):
         assert event_loop.steps == [1]*10
 
 
+@pytest.mark.asyncio
+async def test_until(assert_run, event_loop):
+    with event_loop.assert_cleanup():
+        xs = (stream.range(1, 10)
+              | add_resource.pipe(1)
+              | pipe.until(lambda x: x == 3))
+        await assert_run(xs, [1, 2, 3])
+
+    async def afunc(x):
+        await asyncio.sleep(1)
+        return x == 3
+
+    with event_loop.assert_cleanup():
+        xs = (stream.range(1, 10)
+              | add_resource.pipe(1)
+              | pipe.until(afunc))
+        await assert_run(xs, [1, 2, 3])
+        assert event_loop.steps == [1] * 4
+
+
 @pytest.mark.asyncio
 async def test_takewhile(assert_run, event_loop):
     with event_loop.assert_cleanup():