blackbull.mqtt.tap

blackbull.mqtt.tap

Best-effort application taps independent of broker routing.

Actor mode drops newest on bounded-inbox overflow; inline mode backpressures only the publishing connection. Both share matching. A {name} topic segment captures one level and supplies a keyword argument.

Message dataclass

Immutable published-message view for on_message taps.

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.

TapDeliver dataclass

Bases: Message

Hand a published Message to the TapActor.

compile_tap(topic, callback)

Compile a topic filter (possibly with {name} captures) into a Tap.

compile_taps(handlers)

Normalise a handler list to Tap objects.

Accepts already-compiled Tap objects or (topic, callback) pairs, so direct callers (tests, benchmarks) can keep the lightweight tuple form.

run_taps(taps, message, *, raise_exceptions=False) async

Invoke every tap whose filter matches message (sequential, isolated).

Captured {name} segments are passed as keyword arguments; a handler without captures is simply called callback(message).

A handler exception is logged and isolated by default — taps are best-effort observers, and one raising handler must not stop the others (nor the broker). raise_exceptions=True propagates the first one instead, so a failing tap fails a test rather than a log line.