moat-lib-cmd 0.3.3

Rationale

MoaT contains some components which require a possibly bidirectional stream of asynchronous messaging, including request/reply interactions and data streaming.

This library supports such interactions.

Prerequisites

The library requires a reliable underlying transport. MoaT uses CBOR, but any reliable, non-reordering messsage stream that can encode basic Python data structures (plus whatever objects you send/receive) works.

The MoaT-Cmd library does not itself call the transport. Instead it affords basic async methods to iterate on messages to send, and to feed incoming lower-level data in.

Usage

from moat.util import packer, stream_unpacker

async def handle_command(msg):
    if msg.cmd[0] == "Start":
        return "OK starting"

    if msg.cmd[0] == "gimme data":
        async with msg.stream_w("Start") as st:
            for i in range(10):
                await st.send(i+msg.data["x"])
            return "OK I'm done"

    if msg.cmd[0] == "alive":
        async with msg.stream_r("Start") as st:
            async for data in st:
                print("We got", data)
        return "OK nice"

    raise ValueError(f"Unknown: {msg !r}")

async with Transport(handle_command) as tr, anyio.create_task_group() as tg:
    decoder = stream_unpacker(cbor=True)

    def reader():
        # receive messages from channel
        async for msg in channel.receive():
            decoder.feed(msg)
            for m in decoder:
                await tr.msg_in(m)

    def sender():
        # send messages to channel
        while True:
            msg = await tr.msg_out()
            await channel.send(packer(msg))

    def request():
        # streaming data in
        msg = await tr.cmd("Start", x=123)
        print("Start", msg)
        async with tr.stream_r("gimme data") as st:
            print("They are starting", st.msg)
            async for msg in st:
                print("I got", msg)
        print("They are done", st.msg)
        # may be None if they didn't send a stream

    def int_stream():
        # streaming data out
        async with tr.stream_w("alive") as st:
            print("They replied", st.msg)
            i = 0
            while i < 100:
                await st.send(i)
                i += 1
                anyio.sleep(1/10)
            st.msg = "The end."
        print("I am done", st.msg)


    tg.start_soon(reader)
    tg.start_soon(sender)
    tg.start_soon(handler)

    tg.start_soon(request)
    tg.start_soon(int_stream)

Specification

All MoaT-Cmd messages are non-empty lists whose first element is a small integer, identifying a sub-channel. Messages that don’t match this description MAY be used for out-of-band communication.

A transport that enforces message boundaries MAY send each message without the leading array mark byte(s).

MoaT-Cmd messaging is simple by design and basically consists of a command (sent from A to B) followed by a reply (sent from B to A). Both directions may independently indicate that more, streamed data will follow. The first and last message of a streamed command or reply are considered to be out-of-band.

The side opening a sub-channel uses non-negative integers as channel ID. Replies carry the ID’s bitwise-negated value. Thus the ID spaces of both directions are separate.

IDs are allocated with the first message on a sub-channel. They MUST NOT be reused until final messages have been exchanged. Exactly one final message MUST be sent in both directions.

Examples