Idea: Timeouts on chunks about aiostream HOT 4 CLOSED

@operator(pipable=True)
async def taketimeout(source, n, timeout=1):
    """Forward the first ``n`` elements from an asynchronous sequence.
    Additionally provide a ``timeout`` in seconds. Default is 1 second.

    If ``n`` is negative, it simply terminates before iterating the source.
    """
    event = asyncio.Event()
    async def timer(event: asyncio.Event, timeout):
        await asyncio.sleep(timeout)
        event.set()
    asyncio.create_task(timer(event, timeout))
    async for item in take(source, n):
        if event.is_set():
            break
        yield item

@pytest.mark.asyncio
async def test_taketimeout(assert_run, event_loop):
    with event_loop.assert_cleanup():
        xs = stream.count(interval=1) | add_resource.pipe(1) | pipe.taketimeout(6, 3)
        await assert_run(xs, [0, 1, 2])

    with event_loop.assert_cleanup():
        xs = stream.count(1) | add_resource.pipe(1) | pipe.taketimeout(0)
        await assert_run(xs, [])

import asyncio
from aiostream import stream, pipe


async def main():
    xs = (
        stream.count(interval=1)  # Count from zero every 1s from an unbound infinite generator
        | pipe.timeout(10) # Close the pipeline after 10 seconds yielding [0,1,2,3,4,5,6,7,8,9]
        | pipe.skip(2)  # Skip the first 2 numbers yielding [2,3,4,5,6,7,8,9]
        | pipe.take(5)  # Take the following 2 yielding [2,3,4,5,6]
        | pipe.filter(lambda x: x % 2)  # Keep odd numbers yielding [3,5]
        | pipe.map(lambda x: x ** 2)  # Square the results yielding [9,25]
        | pipe.accumulate()  # Add the numbers together yielding 34
    )
    print(await xs)


# Run main coroutine
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

import random
from contextlib import asynccontextmanager

import asyncio
from aiostream import stream, pipe, operator, streamcontext


@asynccontextmanager
async def buffer(streamer, size=1):
    queue = asyncio.Queue(maxsize=size)
    sentinel = object()

    async def consume():
        try:
            async for item in streamer:
                await queue.put(item)
        finally:
            await queue.put(sentinel)

    @operator
    async def wrapper():
        while True:
            item = await queue.get()
            if item is sentinel:
                await future
                return
            yield item

    future = asyncio.ensure_future(consume())
    try:
        yield wrapper()
    finally:
        future.cancel()


@operator(pipable=True)
async def catch(source, exc_cls):
    async with streamcontext(source) as streamer:
        try:
            async for item in streamer:
                yield item
        except exc_cls:
            return


@operator(pipable=True)
async def chunks(source, n, timeout):
    async with streamcontext(source) as streamer:
        async with buffer(streamer) as buffered:
            async with streamcontext(buffered) as first_streamer:
                async for first in first_streamer:
                    tail = await (
                        buffered
                        | pipe.timeout(timeout)
                        | catch.pipe(asyncio.TimeoutError)
                        | pipe.take(n - 1)
                        | pipe.list()
                    )
                    yield [first, *tail]


async def random_sleep(item):
    return await asyncio.sleep(random.random() * 0.15, result=item)


async def main():
    xs = stream.count() | pipe.map(random_sleep, task_limit=1)
    ys = xs | chunks.pipe(5, timeout=0.1) | pipe.print()
    await ys


if __name__ == "__main__":
    asyncio.run(main())

async def main():
    source=stream.range(0,100)
    even=source|pipe.filter(lambda x:x%2==0)
    xs=even|chunks.pipe(9,0.1)|pipe.map(lambda x:f'even {x}',task_limit=1)
    await (xs|pipe.print())
if __name__ == "__main__":
    asyncio.run(main())