在 python3 中合并异步迭代

有没有一个好的方法，或者一个支持良好的库，用于在 python3 中合并异步迭代器？

期望的行为与在reactivex中合并可观察量的行为基本相同。

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

（来源：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())

您可以使用aiostream.stream.merge:

from aiostream import stream

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

更多示例在文档和这个answer.