Stream Fork about aioreactive HOT 6 CLOSED

To split a stream in to multiple streams:

from aioreactive.producer import Producer
import aioreactive.producer.operators as ops

xs = Producer.from_iterable(range(10))
odds = xs | ops.filter(lambda x % 2 == 1)
evens = xs | ops.filter(lambda x: x % 2 == 0)
...

Perfectly ok to ask questions in issues.

from aioreactive.

from aioreactive.

Have a look at this example.

from aioreactive.

Ha cool, that's exactly what I ended up doing. But a side question, If I have a Producer that comes from an iterator, which is essentially reading an async response. If I fork like this, it seems that each instance of the stream with execute the request and get subsequent values from that. I guess this is the expected behavior right?

from aioreactive.

Yes, that is expected behavior. This is what's called "cold" observables in the Rx world. You can turn them "hot" by publishing. Not a 'publish()' operator in aioreactive yet, but you can simulate it by using an AsyncStream as your sink. Thus you split from the AsyncStream, then you start the from_iterable() using your stream instead of the sink (a stream is also a sink).

Something like:

xs = Producer.from_iterable()

ys = AsyncStream()
odds = ys | ops.filter(lambda x: x % 2 == 1)
evens = ys | ops.filter(lambda x: x % 2 == 0)

await start(odds, FuncSink(mysink))
await start(evens, FuncSink(mysink))
await start(xs, ys)

from aioreactive.

Ha, cool. I guess an operator like that would have to mimic what the AsyncStreamFactory does for the start? Create the intermediate stream and connect them up together?

from aioreactive.