blackbull.mqtt.connection

blackbull.mqtt.connection

MQTT reader and sole-writer connection actor.

All output traverses the writer inbox; Close changes terminal state without requiring a queue slot. Supervise both tasks and finish Detach before teardown.

MQTT5Actor

Bases: Actor

One per MQTT 5.0 connection; the sole writer to its socket.

Tap dispatch is selected at construction: pass a running TapActor as tap for decoupled (actor-mode) dispatch, or app_handlers for inline dispatch on this connection.

read_loop(reader) async

Decode the wire and forward control packets to the broker.

Reads block on a real socket; a fake reader may return b'' while merely idle, in which case we poll. Either way the loop ends on EOF (at_eof()), a DISCONNECT (sets _done), or task cancellation.

PacketFramer

Decode packets at strict wire boundaries, retaining incomplete input.

Malformed packets raise MQTTDecodeError; callers must end the stream. max_packet_size counts all wire bytes, including the fixed header; zero disables the cap. Oversized declarations raise PacketTooLarge as soon as the Remaining Length is complete.

buffered property

Bytes held for the next feed — the thing a size bound must keep small.

PacketTooLarge

Bases: Exception

A packet declared more bytes than BB_MQTT_MAX_PACKET_SIZE allows.

Carries the declared size rather than a buffered one: the whole point of the check is that the payload is refused on the strength of the fixed header, so there is nothing buffered to report.

serve_connection(reader, writer, ctx, broker, *, app_handlers=None, tap=None) async

Raw-protocol handler body for one MQTT connection.

Spawns the connection actor's inbox-drain (run) alongside the reader loop, and guarantees the broker sees a Detach when the connection ends — so a Will fires on an abnormal (cancelled) close. Pass tap for decoupled tap dispatch or app_handlers for inline dispatch (see blackbull.mqtt.tap).