How to dispatch items from a single source to several concurrent streams? about aiostream HOT 2 OPEN

Hey! Thanks for the report :)

I am trying to understand the meaning of 'a stream can be streamed multiple times'.

Oh yea that can be confusing. That simply means that a given stream can be used multiple times, as in:

    xs = stream.range(3) | pipe.list()
    assert await xs == [0, 1, 2]
    assert await xs == [0, 1, 2]

And I can confirm it is fine to run them concurrently, as they correspond to two different streamer instantiation:

    assert await asyncio.gather(xs, xs) == [[0, 1, 2], [0, 1, 2]]

Your example is a bit trickier though, as the stream you built depends on an external source, which means the two streamers iterate the same async generator concurrently. Here's how I would re-write it:

import asyncio
from functools import partial
from aiostream import stream, pipe, async_


async def produce():
    i = 1
    while True:
        yield i
        i += 1
        await asyncio.sleep(1)


async def consume(cid, item):
    print(f'Consumer {cid} got: {item}')
    await asyncio.sleep(.1)
    return item


async def main():
    producer_stream = stream.iterate(produce())
    async with producer_stream.stream() as producer:
        consumer_stream_1 = (
            stream.preserve(producer)
            | pipe.take(2)
            | pipe.map(async_(partial(consume, 1)))

        )
        consumer_stream_2 = (
            stream.preserve(producer)
            | pipe.take(4)
            | pipe.map(async_(partial(consume, 2)))
        )
        await asyncio.gather(consumer_stream_1, consumer_stream_2)


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

Notice how stream.preserve is used to prevent consumer_stream_1 from closing the producer once it's finished so consumer_stream_2 can keep working on it.

I noticed 2 bugs while playing with this example plus the one you already noticed on the anyio branch, I'll try to do some proper reporting tomorrow.

Let me know if you have any questions :)

from aiostream.

After consumer 1 finishes, this crashes for me with:

AttributeError: 'generator' object has no attribute 'cr_await'

But if I understand you correctly, there shouldn't be anything inherently wrong with trying to iterate the same external generator in parallel?

from aiostream.