`blackbull.mqtt.extension`¶

`blackbull.mqtt.extension` ¶

User-facing wiring for the MQTT 5 broker — detector + MQTTExtension.

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

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

blackbull.mqtt.extension¶