在python3中合并异步可迭代

Question

有没有很好的方法或受支持的库在python3中合并异步迭代器？

所需的行为与在reactx中合并可观察对象的行为基本相同。

也就是说，在正常情况下，如果要合并两个异步迭代器，则希望生成的异步迭代器按时间顺序产生结果。迭代器之一中的错误应使合并的迭代器出轨。

（来源：http : //reactivex.io/documentation/operators/merge.html）

这是我的最佳尝试，但似乎可能存在以下标准解决方案：

async def drain(stream, q, sentinal=None):
    try:
        async for item in stream:
            await q.put(item)
        if sentinal:
            await q.put(sentinal)
    except BaseException as e:
        await q.put(e)


async def merge(*streams):

    q = asyncio.Queue()
    sentinal = namedtuple("QueueClosed", ["truthy"])(True)

    futures = {
        asyncio.ensure_future(drain(stream, q, sentinal)) for stream in streams
    }

    remaining = len(streams)
    while remaining > 0:
        result = await q.get()
        if result is sentinal:
            remaining -= 1
            continue
        if isinstance(result, BaseException):
            raise result
        yield result


if __name__ == "__main__":

    # Example: Should print:
    #   1
    #   2
    #   3
    #   4

    loop = asyncio.get_event_loop()

    async def gen():
        yield 1
        await asyncio.sleep(1.5)
        yield 3

    async def gen2():
        await asyncio.sleep(1)
        yield 2
        await asyncio.sleep(1)
        yield 4

    async def go():
        async for x in merge(gen(), gen2()):
            print(x)

    loop.run_until_complete(go())

Answer 1

您可以使用aiostream.stream.merge：

from aiostream import stream

async def go():
    async for x in stream.merge(gen(), gen2()):
        print(x)

文档中的更多示例以及此答案。