message-channel

此库提供了一个消息通道对象，它从消息集合中减去特定的消息。它类似于SQL或ReactiveX的group by，但适用于异步读取器。

安装

pip install python-message-channel

使用方法

例如，假设您有一个字符串流，其中消息以a、b、... e等前缀开头，并且您希望为以b或d等前缀开头的消息分割子通道，如下所示。

=============================================
---------------------------------> a:foo
--------------------+
--------------------|------------> c:foo
--------------------|------------> d:foo
--------------------|------------> e:foo
====================|========================
channel             |
                   =|========================
                    +------------> b:foo
                   ==========================
                   subchannel `m.startswith('b')`

此库是处理此类情况的工具。首先，从一个流读取器创建一个Channel实例，然后您可以通过channel.recv()方法接收消息。在本例中，我们使用asyncio.Queue作为流。

import asyncio

from message_channel import Channel

async def main():
    # Create original stream
    stream = asyncio.Queue()

    # Create stream reader
    async def reader():
        return await stream.get()

    # Create stream channel
    async with Channel(reader) as channel:
        stream.put_nowait('a:foo')
        stream.put_nowait('b:foo')
        stream.put_nowait('c:foo')
        stream.put_nowait('d:foo')
        stream.put_nowait('e:foo')
        assert (await channel.recv()) == 'a:foo'
        assert (await channel.recv()) == 'b:foo'
        assert (await channel.recv()) == 'c:foo'
        assert (await channel.recv()) == 'd:foo'
        assert (await channel.recv()) == 'e:foo'


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

您可以通过一个预测器使用channel.split()方法分割通道。

    async with Channel(reader) as channel:
        def predicator(m):
            return m.startswith('b:')

        async with channel.split(predicator) as sub:
            stream.put_nowait('a:foo')
            stream.put_nowait('b:foo')
            stream.put_nowait('c:foo')
            stream.put_nowait('d:foo')
            stream.put_nowait('e:foo')
            # sub receive messages starts from 'b:'
            assert (await sub.recv()) == 'b:foo'
            # channel (original) receive messages other than above
            assert (await channel.recv()) == 'a:foo'
            assert (await channel.recv()) == 'c:foo'
            assert (await channel.recv()) == 'd:foo'
            assert (await channel.recv()) == 'e:foo'

API文档

https://fixpoint.github.io/python-message-channel/

由pdoc提供支持。

许可证

在MIT许可证下分发