MQTT 5 broker

BlackBull ships a pure-Python MQTT 5 broker that runs as a sidecar on the Non-ASGI bridge. One process can serve HTTP/1.1, HTTP/2, and WebSocket and speak MQTT on the standard 1883 port — no separate broker, no C extension, no extra dependency.

from blackbull import BlackBull
from blackbull.mqtt import MQTTExtension, Message

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

@app.route(path='/')
async def index():
    return "HTTP here; MQTT broker on :1883."

@mqtt.on_message(topic='sensors/+/temperature')
async def on_temperature(msg: Message):
    print(msg.topic, msg.payload.decode())

app.run(port=8000)   # HTTP on 8000, MQTT on 1883

A full runnable version is in examples/mqtt_broker.py.

Wiring and the handler API

The broker is an Extension: you register it through the core's single extension seam, app.add_extension(MQTTExtension(port=1883)), which returns the extension so you can keep a handle on it. BlackBull itself carries no MQTT-specific code.

MQTTExtension.on_message(topic='#') decorates an async (message) -> None callback. It receives a single blackbull.mqtt.Message (msg.topic, msg.payload, msg.qos, msg.retain, msg.properties) — mirroring how @app.on hands an observer one Event. The callback fires for every PUBLISH whose topic matches topic — an MQTT topic filter, so the + (single level) and # (multi level) wildcards apply.

mqtt = app.add_extension(MQTTExtension())

@mqtt.on_message(topic='#')            # every message
async def firehose(msg: Message): ...

@mqtt.on_message(topic='alerts/#')     # one subtree
async def alerts(msg: Message): ...

A filter level may also be a {name} capture: it matches one level like + and is injected into the callback as a keyword argument, mirroring HTTP path params.

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

Handlers are an application-level tap: they run in addition to normal broker routing, never instead of it. The broker still delivers each PUBLISH to every subscribed MQTT client whether or not a handler matches. A handler that raises is isolated and logged — it never disturbs the broker or other handlers.

By default taps are dispatched on a decoupled TapActor: the connection hands each message off without waiting, so a slow tap can never back-pressure delivery or the broker. The TapActor's inbox is bounded; if taps fall behind, the newest messages are dropped (best-effort observability) and a running dropped-count is logged. Taps are therefore not a reliable delivery path — use a real MQTT subscription for that. (MQTTExtension(tap_mode='inline') runs taps inline on the receiving connection instead, so slow taps delay that reader.)

The broker also runs without any handler at all: on_message is just how an application observes traffic. app.add_extension(MQTTExtension()) on its own gives you a fully functional broker with no tap.

Tap dispatch waits for the broker to confirm that the publishing connection is admitted at that command's position in its inbox. Rejected or retired connections cannot trigger taps through queued PUBLISH packets. This confirms connection admission, not successful routing, retained storage, validation of the PUBLISH, or a subscriber's acknowledgement. Actor-mode callbacks remain decoupled; inline callbacks run on the reader, never inside the broker or the sole writer. Cancellation and broker shutdown can omit a pending observation.

Connection admission and retirement

CONNECT is allowed once per network connection. Clients may pipeline control packets immediately after it without waiting for CONNACK; the broker processes that FIFO in order and sends a successful CONNACK before other normal replies, including PINGRESP. A rejected CONNECT does not authorize later PUBLISH, subscription changes or QoS acknowledgements. AUTH's exception in MQTT 5 §3.1.4 does not grant admission or reopen a rejected connection; this broker does not implement an enhanced-authentication negotiation.

A duplicate CONNECT on an admitted connection is a Protocol Error (0x82). Server-only packet types received from the client also end admission instead of being skipped; a non-CONNECT first command cannot establish a connection. Reconnecting requires a new network connection, which can resume an existing session. Session takeover invalidates the old actor, so its delayed commands cannot publish or alter the replacement session, even while its writer flushes. A zero-length Client Identifier never takes part in that: the broker assigns an unused identifier and reports it in CONNACK. Every CONNACK honours the peer's Maximum Packet Size: the full reply, then one trimmed of the broker's advertised limits (an assigned identifier is not optional), and only when neither fits is the connection refused with 0x95 — or closed outright when even that reply cannot fit — before creating a session.

The limit also applies to outgoing PUBLISH and PUBREL packets. A packet that exceeds it is discarded, and its delivery is treated as complete [MQTT-3.1.2-25]. It is never queued or retried. The limit belongs to the connection: queued messages are rechecked against the reconnecting client's limit. Retained and Will messages follow the same rule. For a Shared Subscription, the message goes to a member that can receive it and is dropped only if no member can. The takeover (0x8E) and duplicate-CONNECT (0x82) DISCONNECTs are not size-checked.

Broker-initiated protocol-error closure retires admission before processing the next command. Client DISCONNECT and transport failures use the FIFO Detach boundary: commands ordered before it were admitted while the connection was active; commands after it cannot mutate broker state. Will and session-expiry processing belong to that retirement, not to a delayed second teardown.

Client identifiers alone are not admission credentials. This protocol sequencing is not application authentication or topic authorization.

What the broker implements

The broker targets the MQTT 5.0 OASIS feature set exercised by BlackBull's conformance matrix:

Area	Support
Connection	CONNECT / CONNACK, protocol-level check (rejects non-5 with 0x84), Clean Start, Session Present
Subscriptions	SUBSCRIBE / SUBACK, UNSUBSCRIBE / UNSUBACK, + and # wildcards, $-topic rules; No Local, Retain As Published, and Retain Handling (0/1/2, §3.3.1.3) options honoured; invalid Topic Filters rejected (0x8F)
Shared subscriptions	$share/{ShareName}/{filter} groups (§4.8.2): each matching message goes to exactly one connected group member, round-robin — see below
Publish	invalid Topic Names (wildcards / null) rejected (0x90), never routed or retained
QoS 0	fire-and-forget delivery
QoS 1	PUBACK round-trip
QoS 2	PUBREC → PUBREL → PUBCOMP four-way handshake; duplicate PUBLISH de-duplicated; unacked messages replayed with DUP=1 on reconnect
Retained	one retained message per topic; delivered to late subscribers; zero-length payload clears
Will (LWT)	delivered on abnormal disconnect; suppressed on a normal DISCONNECT (0x00)
Keep-alive	PINGREQ / PINGRESP; an idle connection is closed (Will fired) at 1.5× the negotiated Keep Alive (§3.1.2.10)
Session takeover	a second CONNECT for a live Client Identifier disconnects the prior connection with 0x8E (§3.1.4)
Client Identifier	a zero-length Client Identifier is assigned an identifier no live or offline session holds, returned in CONNACK as Assigned Client Identifier (§3.2.2.3.7)
Properties	the full MQTT 5 property set (§2.2.2.2) on every packet that carries properties
Sessions	subscriptions and pending QoS state preserved across reconnects with Clean Start = 0
Flow control	the client's Receive Maximum (§3.1.2.11.3) is enforced in the outbound direction, per Network Connection; the broker's own is advertised in CONNACK as a promise to conforming clients
Resource limits	packet size, session backlog, subscription and session counts, and retained-store size are bounded; the ones MQTT 5 has a property for are advertised — see below

Variable Byte Integers accept non-minimal encodings: a value whose encoding carries a redundant continuation octet (0 in two octets, 300 in three) decodes normally. A non-minimal encoding violates [MQTT-1.5.5-1] ("the minimum number of bytes necessary"); by the spec's definition such a packet is a Malformed Packet (Terminology; §4.13.1). This runtime does not enforce that. The tolerance is pinned in tests/unit/test_mqtt_codec_lengths.py.

Shared subscriptions

Subscribers that name the same $share/{ShareName}/{filter} pair form a share group (MQTT 5 §4.8.2). Each application message matching the filter is delivered to exactly one member of the group — round-robin across the members currently connected — instead of every matching subscriber. This is the standard MQTT pattern for load-balancing a work queue across a pool of consumers:

# terminals 1 and 2 — two workers in the same share group
mosquitto_sub -t '$share/pool/jobs/#' -p 1883 -V 5
mosquitto_sub -t '$share/pool/jobs/#' -p 1883 -V 5

# terminal 3 — publishes alternate between the two workers
mosquitto_pub -t 'jobs/import' -m 'job-1' -p 1883 -V 5
mosquitto_pub -t 'jobs/import' -m 'job-2' -p 1883 -V 5

Semantics worth knowing:

• Non-shared subscriptions are unaffected — a client subscribed to a plain jobs/# still receives every message, alongside whichever group member the rotation picks. Shared and non-shared subscriptions held by the same client are independent delivery channels.
• Delivery QoS is the chosen member's granted QoS (capped by the publish QoS, as usual).
• Disconnected members are skipped while any member is connected (§4.8.2.3). If no member is connected, the message is not queued for the group — the same no-offline-queue behaviour the broker applies to non-shared subscriptions.
• Retained messages are never delivered to a shared subscription (§4.8.2).
• No Local cannot be combined with a shared subscription — MQTT 5 makes it a Protocol Error (§3.8.3.1), and the broker disconnects with 0x82.
• Malformed forms ($share/g, an empty ShareName, a wildcard in the ShareName, or an empty filter portion) are rejected per-entry with 0x8F.

Resource limits

An MQTT broker holds state on a client's behalf: buffered packet bytes, unacked messages, retained messages that outlive the session that published them. Each of those is bounded, and each bound is advertised in CONNACK where MQTT 5 has a property for it. Local aggregate quotas can also be reached by legal traffic, for example when a subscriber's network slows down.

DISCONNECT refusals below apply after CONNECT admission. Earlier packet-size or input-budget refusals close the connection silently.

Limit	Default	Advertised as	Over the limit
BB_MQTT_MAX_PACKET_SIZE	1 MiB	Maximum Packet Size (§3.2.2.3.6)	DISCONNECT 0x95 Packet Too Large, connection closed
BB_MQTT_RECEIVE_MAXIMUM	64	Receive Maximum (§3.2.2.3.3)	a conforming client waits — a promise, not a gate (see below)
BB_MQTT_MAX_QUEUED_MESSAGES	1000	— (a broker-side total)	newest message refused, cap hit logged
BB_MQTT_BROKER_INBOX_MAXSIZE / BB_MQTT_BROKER_INBOX_MAX_BYTES	1024 messages / 16 MiB	— (worker input handoff)	readers await admission; a single packet exceeding the byte budget is refused with 0x97
BB_MQTT_CONNECTION_INBOX_MAXSIZE / BB_MQTT_CONNECTION_INBOX_MAX_BYTES	1024 packets / 16 MiB	— (per-connection output handoff)	after giving the writer a scheduling opportunity, terminate only the connection whose queue cannot admit the packet
BB_MQTT_MAX_RETAINED	10000	— (a broker-side total)	retained publish to a new topic refused; 0x97 in the PUBACK/PUBREC at QoS ≥ 1
BB_MQTT_MAX_SUBSCRIPTIONS	1000	— (a per-session unit)	a new Topic Filter refused with 0x97 in the SUBACK; re-subscribing to one the session holds always works
BB_MQTT_MAX_SESSIONS	10000	— (a broker-side total)	CONNECT for an unknown Client Identifier refused with 0x97 in the CONNACK, connection closed; a resuming client is admitted

These properties of the limits are worth knowing before you tune them:

Actor mailboxes and the QoS backlog are different owners. The broker inbox holds decoded CONNECT, PUBLISH and subscription changes, plus compact ACK and lifecycle messages. Each connection's writer inbox holds all packet types, including QoS 0, retained replay, shared-subscription deliveries and control replies. Both inboxes have independent positive count and byte limits; invalid environment values (including zero) fall back to the default. Their defaults allow a full-size packet accepted by the default packet limit, and a burst larger than the count limit can proceed while a healthy consumer drains it. If you raise the packet limit, size the mailbox byte budgets for those packets too. Reducing a mailbox budget below one accepted packet can refuse that legal packet even when the mailbox is empty.

The byte charge is encoded wire size, not Python heap usage. The broker charges the retained packet's wire size (compact ACK/lifecycle envelopes charge one byte and one message slot); output charges the encoded packet bytes and also retains its packet object until the write finishes. Each consumer can hold one active message outside its inbox; each blocked reader can hold one admission candidate and its framer buffer. The server's connection limit bounds the number of readers and per-connection inboxes. These budgets do not include session pending QoS state, held QoS messages, retained storage or the tap inbox. Those retain their own count/packet-size limits and expiry or shutdown owners.

A slow output socket does not stall broker routing for other clients. Output admission failure logs a cap hit and ends that connection; it cannot promise a DISCONNECT reason on the wire. Queued QoS 0 is best-effort. Active writes and close flushing use BB_WRITE_TIMEOUT, which zero disables.

The packet limit includes the fixed header and is checked as soon as the header is complete; zero disables it. Packet decoding errors or oversized declarations end the connection without forwarding trailing packets. Malformed input closes silently before CONNECT admission; afterward it receives DISCONNECT (0x81). Keep Alive bounds receive idleness after CONNECT, not total packet-assembly time; zero disables that idle check.

The QoS backlog keeps older promised messages and refuses the newest when full. It is separate from writer mailboxes and retained storage.

The send quota belongs to the connection, and retransmissions spend it too. Every CONNECT re-declares the client's Receive Maximum; an omitted property means 65535, and a declared zero is a Protocol Error (0x82). That window is shared by live deliveries and the DUP=1 retransmissions of a resumed session (§4.4), which go first, in the order the originals were sent (MQTT-4.4.0-2). A matching PUBACK, PUBCOMP, or error PUBREC frees one slot; a successful PUBREC keeps its slot until PUBCOMP. How many unacknowledged messages the session holds is never the quota. If all 65535 packet identifiers are still in use, new QoS>0 deliveries wait in the bounded outbound queue until an exchange completes, even when this connection has quota available. Control packets (PUBREL and the acknowledgements of the client's own publishes) flow even when the window is full.

Retained messages are capped by topic count, and correction is always allowed. At the cap, a retained publish to a new topic is refused, but updating or deleting an already-retained topic still works. A client locked out of correcting its own retained state would be worse off than one that could never set it — and deleting (a zero-length retained payload, §3.3.2.3) is what frees the room being contended for. The message is still delivered to current subscribers; only the storage is declined.

How the publisher finds out depends on the QoS it chose, because that is what decides whether the protocol has a channel for the answer. QoS 1 and 2 receive 0x97 (Quota Exceeded) in the PUBACK or PUBREC. QoS 0 is not told — it has no acknowledgement (§3.3.4), and closing the connection over a storage quota would be disproportionate as well as destroying a live delivery that succeeded. If you need to know your retained state was stored, publish it at QoS ≥ 1.

Session state is bounded on all three axes, and one of them is the client's to choose. How big one session may be is BB_MQTT_MAX_SUBSCRIPTIONS plus BB_MQTT_MAX_QUEUED_MESSAGES; how many may exist is BB_MQTT_MAX_SESSIONS; how long a session outlives its connection is the Session Expiry Interval the client declares in CONNECT (and may shorten in DISCONNECT, §3.14.2.2.2). The third is the one you do not control — and §3.1.2.11.2 lets a client declare 0xFFFFFFFF, meaning never expires. That is legal, BlackBull honours it, and it is exactly why the total exists: a peer cycling Client Identifiers with a never-expiring interval is bounded by BB_MQTT_MAX_SESSIONS and by nothing else. At that cap a CONNECT for an unknown Client Identifier is refused with 0x97; a client resuming a session already in the table is admitted, because refusing it would free nothing.

One limit is advertised but not enforced, and it is worth knowing which. BB_MQTT_RECEIVE_MAXIMUM tells a client how many QoS>0 publishes it may have in flight towards the broker. Nothing counts a non-conforming client's against it; what bounds that direction is the 16-bit packet-identifier space and the packet size cap. The client's Receive Maximum, in the outbound direction, is enforced — that is what BB_MQTT_MAX_QUEUED_MESSAGES backs.

Every refusal above emits a record on blackbull.caps (see Logging), so a limit that fires is a limit you can see fire.

Trying it with Mosquitto

The broker speaks standard MQTT 5, so the Eclipse Mosquitto CLI works against it (apt install mosquitto-clients):

# terminal 1 — subscribe
mosquitto_sub -t 'sensors/#' -p 1883 -V 5

# terminal 2 — publish
mosquitto_pub -t 'sensors/room1/temperature' -m '21.5' -p 1883 -V 5

The message appears in the subscriber's terminal and in any matching @mqtt.on_message handler.

Testing the taps

Standing up a real broker and a real MQTT client to exercise a @mqtt.on_message handler is heavy for a unit or integration test. MQTTTestBroker feeds PUBLISHes into the app's registered taps directly — no socket, no CONNECT, no client:

from blackbull import BlackBull
from blackbull.mqtt import MQTTExtension, Message
from blackbull.testing.mqtt import MQTTTestBroker

app = BlackBull()
mqtt = app.add_extension(MQTTExtension(port=1883))
captured = []

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

async def test_tap():
    async with MQTTTestBroker(app) as broker:
        await broker.publish(topic='sensors/room1/temperature', payload=b'21.5')
    assert captured == [('room1', b'21.5')]

publish matches the topic against the registered filters — + / # wildcards and {name} captures included — runs the matching taps, and awaits them, so the assertion can run the moment publish returns. A tap that raises propagates its exception out of publish: where production logs and isolates a failing tap (taps are best-effort observers), a test wants the failure visible, so the exception surfaces and the remaining taps in that publish are skipped.

No broker needs to be running and no socket is bound — the helper is async-only, feeding the taps on the test's own event loop. Only the tap pipeline is exercised: broker routing, QoS flows and retained messages are the conformance suite's territory, not this helper's.

Documenting the taps with AsyncAPI

OpenAPI documents BlackBull's HTTP surface, but it has no vocabulary for topics or the publish/subscribe direction, so the broker is invisible to it. The messaging-world counterpart is AsyncAPI, and AsyncAPIExtension emits an AsyncAPI 3.0 document for the topic taps your app registered — served over HTTP, exactly as /openapi.json is. It is a normal extension and coexists with OpenAPIExtension:

from blackbull import BlackBull
from blackbull.mqtt import MQTTExtension, AsyncAPIExtension, Message

app = BlackBull()
mqtt = app.add_extension(MQTTExtension(port=1883))
app.add_extension(AsyncAPIExtension(title='Sensor Gateway', version='1.0.0'))

@mqtt.on_message(topic='sensors/{room}/temperature')
async def on_temp(msg: Message, room: str):
    """Temperature readings per room."""

After app.run() the document is at /asyncapi.json and an HTML viewer (a CDN-hosted AsyncAPI renderer — no new Python dependency) at /asyncapi. Each on_message filter becomes a channel whose address is the filter as you wrote it ({name} captures preserved); each callback becomes a receive operation (the application receives PUBLISHes), with its docstring summary. Pass docs_path=None to skip the HTML page, or server_host= to override the advertised broker host (default localhost:<port>).

The document is generated lazily on each request, so taps registered after add_extension(AsyncAPIExtension(...)) are still included. The MQTT extension must be present when the spec route is hit, or the request raises RuntimeError.

Three honest caveats — also stated in the document's info.description:

• It documents the application's taps, not "the broker's API". A broker accepts any topic from any client; on_message filters describe only what this app observes.
• QoS and retain are not captured — taps fire regardless of QoS, so MQTT channel bindings are omitted until the tap API carries that metadata.
• Payloads are opaque bytes (application/octet-stream) until a future opt-in schema= on on_message lands.

TLS (mqtts://)

MQTTExtension(port=8883, tls=True) serves the broker port over TLS using the configured server certificate — pass certfile/keyfile to app.run(). The server refuses to start if tls=True is set with no certificate configured. Cleartext remains the default (tls=False), so existing deployments are unchanged.

Why the broker has a single owner

The broker runs on worker 0 only, while HTTP scales across all workers. Its subscriptions, sessions, retained messages and QoS state must remain consistent across connections; this broker has no shared or persistent store.

See Workers for how the single-owner binding interacts with --reload and SO_REUSEPORT.

Limitations

• No MQTT-over-WebSocket transport. TLS is supported via MQTTExtension(tls=True); the WebSocket binding is not yet wired up.
• Single owner (HTTP still scales). The broker runs on worker 0 only — its state (subscriptions, sessions, retained messages) lives in that one process and is neither shared across workers nor persisted across restarts. HTTP, however, scales across all workers: app.run(workers=4) alongside the broker runs HTTP on every worker and the broker on worker 0. (--reload still pins workers=1 when a broker is registered.)
• In-memory sessions. Session state lives in the broker process and does not survive a restart. A session with a finite Session Expiry Interval is removed once the interval elapses (§3.1.2.11.2). 0xFFFFFFFF means does not expire, and BlackBull honours it: such a session is bounded by BB_MQTT_MAX_SESSIONS, not by the clock.