Skip to content

blackbull.mqtt

blackbull.mqtt

MQTT 5 broker through the generic extension mechanism.

Register MQTTExtension with app.add_extension; HTTP dispatch stays independent. See docs/guide/mqtt.md for broker and tap contracts.

AsyncAPIExtension

Bases: Extension

Mount an AsyncAPI 3.0 spec endpoint (+ HTML viewer) describing the MQTT taps.

Requires an MQTTExtension on the same app; it is read lazily (per request) via app.extensions['mqtt'], so there is no ordering constraint at registration — and taps added after this extension are still documented. The MQTT extension must, however, be present when the spec route is hit, or the request raises RuntimeError.

Two GET routes are registered (mirroring OpenAPIExtension):

  • spec_path (default /asyncapi.json) → the document as JSON;
  • docs_path (default /asyncapi, None to skip) → an HTML page hosting the CDN AsyncAPI renderer.

init_app(app)

Wire the spec + docs routes onto app through the public API.

BrokerActor

Bases: Actor

Single owner of all MQTT routing state (one per app/worker).

Connection actors send it client events; it sends Send/Close back to them. Serial inbox processing means no locks and no shared mutable state — the design goal of the actor split.

close()

Release the timer and wake producers whose consumer has stopped.

MQTTExtension

Bases: Extension

Wires the MQTT 5 broker into a BlackBull app as a non-ASGI protocol.

Register it once and tap the broker's routing with on_message::

from blackbull.mqtt import MQTTExtension, Message

mqtt = app.add_extension(MQTTExtension(port=1883))

@mqtt.on_message(topic='sensors/{room}/temperature')
async def on_temp(msg: Message, room: str):
    print(room, msg.payload)

The broker itself (CONNECT/SUBSCRIBE/PUBLISH/QoS flows, retained messages, Will delivery) runs whether or not any handler is registered; handlers are an application-level tap on top of normal broker routing.

This instance owns one BrokerActor and one TapActor, started on app startup and stopped on shutdown — one of each per worker.

tap_mode selects how taps are dispatched: 'actor' (default) runs them on the decoupled TapActor, so a slow tap never back-pressures delivery; 'inline' dispatches on the connection actor itself. tap_queue_size bounds the actor-mode inbox (drop-newest on overflow), and docs/about/mqtt-actor-design.md covers the choice.

__init__(*, port=1883, tls=False, tap_mode='actor', tap_queue_size=1024)

Configure the broker extension.

tap_mode is 'actor' or 'inline' — any other value raises ValueError. The annotation cannot say so, which is why this does.

iter_subscriptions()

Yield a Subscription for each registered on_message tap.

A read-only view of the compiled handlers at call time, so a documentation endpoint calling it per request picks up taps registered after that endpoint was wired in.

iter_taps()

Yield compiled tap registrations present at call time.

on_message(topic='#')

Decorator: register an async (message, **captures) -> None tap for every PUBLISH whose topic matches topic (an MQTT topic filter, so + and # wildcards apply, plus {name} capture segments). The callback receives a Message.

shutdown(app) async

Stop the broker and tap actors on lifespan shutdown.

startup(app) async

Start the broker (and, in actor mode, the tap) inbox loops.

MQTTProtocolDetector

Bases: ProtocolDetector

Recognise an MQTT connection from its first byte.

Every MQTT session opens with a CONNECT packet — fixed-header first byte int(MQTTPacketType.CONNECT) << 4, flags 0. That is unambiguous against the HTTP request line and the HTTP/2 preface, both of which start with ASCII text.

Message dataclass

Immutable published-message view for on_message taps.

Subscription

Bases: NamedTuple

A read-only view of one registered on_message tap.

Yielded by MQTTExtension.iter_subscriptions so documentation tools (AsyncAPIExtension) can describe an app's taps without reaching into the private _handlers list. topic is the filter as the application wrote it ({name} captures restored); callback is the registered coroutine function (its __name__/docstring name the AsyncAPI operation).

Tap dataclass

Compiled on_message filter with named captures.

display_filter property

The topic filter as originally written, with {name} captures restored (match_filter rewrites each to + for matching).

'sensors/{room}/temperature' round-trips back to itself; a plain 'sensors/+/temperature' stays as 'sensors/+/temperature'.

bind(topic)

Return captured {name: value} if topic matches, else None.

TapActor

Bases: Actor

Decoupled, lifespan-owned consumer of on_message taps.

Producers call offer (non-blocking); a single consumer task drains the bounded inbox and runs the matching taps, so FIFO order of accepted messages is preserved and tap latency never reaches the connection or broker.

dropped property

Number of messages dropped on overflow so far (best-effort metric).

offer(message)

Enqueue without blocking; drop newest on overflow and log the dropped count.

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).