Channel and communication

The parent-child channel (client and frame) and the communication mixin.

Communication capability: the first mixin over the base server (D17).

The base server is born WITHOUT channels. This mixin adds the communication capability as member objects built by its cooperative __init__: parent_channel — the ChannelClient of ◆D10, ARMED iff parent=<address> was given — and children_channel — the hub side, a placeholder member the orchestration package arms (the minimal package knows how to BE a child, not how to HAVE children). Accessing an unarmed side raises RuntimeError (“not armed”); a composition WITHOUT the mixin simply lacks the attributes — a different type, not a ghost.

This is the first REAL proof of the D16 cooperative contract: composed BEFORE the server class (class MyServer(CommunicationMixin, BaseServer)) the mixin peels its own kwargs, forwards the rest, and hooks the lifespan without the base knowing it — __call__ cooperatively intercepts the lifespan scope: an armed parent side pre-receives lifespan.startup, connects (and REGISTERs) BEFORE any app hook runs, and disconnects when the protocol completes at shutdown. An unreachable hub answers lifespan.startup.failed and re-raises: a child that cannot register must die, never serve detached.

Registration makes the registrant a child in the tree even when not spawned by us (communication ≠ process lifecycle, D17): the REGISTER frame presents {"name", "pid"}; the name is derived from the class name and the pid (child naming proper belongs to the orchestration package).

class genro_asgi.communication.CommunicationMixin(**kwargs)[source]

Bases: object

Communication capability mixin, composed BEFORE a server class.

Constructor kwargs peeled here: parent — the address of the parent hub (uds:<path> | tcp:<host>:<port>); when given, the parent side is armed with a ChannelClient.

Parameters:: kwargs (Any)

property parent_armed: bool: Whether the parent side was armed (parent= given at init).

property parent_channel: ChannelClient: The channel to the parent hub; unarmed access is an error.

property children_channel: Any: The hub side; a placeholder the orchestration package arms.

Channel client — the child side of the parent↔child channel.

Knowing how to BE a child is part of what a server IS (SPECIFICATION.md ◆D10): the minimal package ships the frame protocol and this client; the hub (parent side) lives in the orchestration package and imports the protocol from below, never the reverse.

connect() retries with short backoff until connect_timeout (boot race: the hub socket may not be bound yet) and presents the child with a REGISTER frame (data={"name", "pid"}). Steady state is fire-and-forget frames in both directions. There is no steady-state reconnection: when the hub side goes away (EOF — the death signal on same-host sockets), on_orphan(client) fires and the child is expected to terminate cleanly. A deliberate close() fires no orphan signal.

Callbacks (on_message(frame), on_orphan(client)) may be sync or async; an exception raised by a callback is logged and never severs the channel — a consumer bug must not fake a member death.

Addresses:

uds:/path/to/hub.sock     Unix domain socket (default)
tcp:127.0.0.1:8731        TCP (multi-host door)

class genro_asgi.channel.client.ChannelClient(address, name, *, on_message=None, on_orphan=None, connect_timeout=10.0, max_size=16777216)[source]

Bases: object

Child-side endpoint: connect to the hub, present itself, relay frames.

Parameters:

address (str)
name (str)
on_message (Callable[..., Any] | None)
on_orphan (Callable[..., Any] | None)
connect_timeout (float)
max_size (int)

property connected: bool: Whether the channel is up (REGISTER sent, receive loop running).

property closed: bool: Whether the channel ended (either side; False before connect).

async connect()[source]

Connect with boot-time retry/backoff, present the REGISTER frame.

Return type:: None

async close()[source]

Deliberate local close: no orphan signal.

Return type:: None

async wait_closed()[source]

Block until the channel ends (either side); the child’s main wait.

Return type:: None

async send(*, method='POST', path='/', data=None)[source]

Send one frame to the hub (fire-and-forget); returns the frame id.

A connection dropping mid-send is a dying hub: the frame is lost by design and the orphan signal follows on the receive side.

Return type:

str

Parameters:

method (str)
path (str)
data (Any)

Frame protocol: length-prefixed wsx envelopes, zero external dependencies.

The wire unit of the channel (SPECIFICATION.md ◆D10: the frame protocol — tiny, zero deps) is a Frame: a wsx envelope (WSX:// prefix + JSON with id/method/path/data, the same payload text the old pool channel carried) preceded by a 4-byte big-endian length. The transport is a plain asyncio stream — the length prefix replaces the websocket framing of the old channel so the minimal package stays dependency-free; the envelope format is unchanged.

FrameStream is the codec over one (StreamReader, StreamWriter) pair: write(frame) sends the wire bytes, read() returns the next Frame or None when the peer is gone — on same-host sockets the channel drops iff a process dies, so EOF is the death signal. Frames are fire-and-forget events (no acks, no replay); the envelope id stays on the wire at zero cost for a future request/response extension.

class genro_asgi.channel.frame.Frame(*, id=None, method='POST', path='/', data=None)[source]

Bases: object

One envelope on the channel (D18: slotted, high cardinality).

An immutable record: id (generated when not given), method, path and data. encode() produces the wire bytes.

Parameters:

id (str | None)
method (str)
path (str)
data (Any)

property id: str: Correlation id (kept on the wire for a future request/response).

property method: str: Envelope method (REGISTER presents a child, POST events).

property path: str: Envelope path, the HTTP-like routing key for the consumer.

property data: Any: JSON-serializable payload (None for no payload).

encode()[source]

The wire bytes: 4-byte big-endian length + WSX:// + JSON.

Return type:: bytes

class genro_asgi.channel.frame.FrameStream(reader, writer, *, max_size=16777216)[source]

Bases: object

Frame codec over one asyncio stream pair (either end of the channel).

read() returns None when the peer is gone (EOF or reset) — the death signal, not an error. An oversized or non-wsx frame is a protocol violation and raises ValueError.

Parameters:

reader (asyncio.StreamReader)
writer (asyncio.StreamWriter)
max_size (int)

async read()[source]

The next frame, or None when the channel ended.

Return type:: Frame | None

async write(frame)[source]

Send one frame; an oversized payload raises ValueError.

Return type:: None
Parameters:: frame (Frame)

async close()[source]

Close the writing side; a peer already gone is not an error.

Return type:: None