Ziplast operator about aiostream HOT 5 CLOSED

Just thought I'd share some ideas for streams that would be useful to me.

Great, thanks!

A ziplast stream

That would be the equivalent of ReactiveX combineLatest right?

Here's a possible implementation:

@operator(pipable=True)
def ziplatest(*sources):
    n = len(sources)

    # Add source index to the items
    new_sources = [
        stream.map(source, lambda x, i=i: {i: x})
        for i, source in enumerate(sources)]

    # Merge the sources
    merged = stream.merge.raw(*new_sources)

    # Accumulate the current state in a dict
    accumulated = stream.accumulate.raw(
        merged, lambda x, e: {**x, **e})

    # Convert the state dict to a tuple
    return stream.map.raw(
        accumulated, lambda x: tuple(map(x.get, range(n))))

And an example:

async def main():
    xs = stream.count(interval=2) | pipe.delay(2)
    ys = stream.count(interval=5) | pipe.delay(5)
    zs = xs | ziplatest.pipe(ys)
    ps = zs | pipe.print()
    await ps

With the corresponding output:

(0, None)
(1, None)
(1, 0)
(2, 0)
(3, 0)
(3, 1)
(4, 1)
(5, 1)
(6, 1)
(6, 2)
(7, 2)
[...]

I guess extra keywords arguments could also be provided, either to disable the producing of incomplete items, or to define a custom default value.

What do you think?

from aiostream.

Awesome! That is exactly what I need.

from aiostream.

Great! I don't really have the time to write the docs and tests for this feature at the moment, so feel free to submit a PR if you're interested. Otherwise, I'll take care of it later.

from aiostream.

Np, thanks for the effort. I really like this library and async generator functions is a pretty nice new python feature. It really cleans up a lot of my previous async context manager code in the project I am currently working at. You get a feeling that async/await is still a pretty young feature in python and there are definitely a few pitfalls to dodge here and there.

from aiostream.

Hi @andersea,

I just added the ziplatest operator (see PR #22), it will become available with version 0.3.1.

Thanks for the suggestion :)

from aiostream.