mqtt-uns-feeder

MQTT / Unified Namespace Feeder

This skill describes how to add MQTT transport to a feeder so that its events are published into a Unified Namespace topic tree carrying binary-mode MQTT 5.0 CloudEvents with JSON payloads. The pegelonline source is the reference implementation — read its files alongside this document.

The skill applies in two shapes:

1. Transport split — an existing source already ships a Kafka feeder and you are adding an MQTT sibling (the pegelonline pattern).
2. MQTT-first — a brand-new source whose data shape is so well-suited to LKV-on-retained-topics that Kafka may not even be added later.

Either way, the runtime never publishes to both transports from the same process. There are always two independent containers, two independent pyproject.toml files, two independent images, and two independent ARM templates.

When to Use

• Add MQTT/UNS publication to an existing Kafka source.
• Greenfield source where the upstream data is naturally a hierarchical catalog of stable identifiers (stations, vessels, zones, sensors) each emitting periodic measurements.
• Refactor a feeder so the acquisition logic is shared between Kafka and MQTT apps.

When not to Use

• High-volume firehose telemetry without stable per-object identity (raw lightning strokes, raw AIS sentences, raw firehoses): UNS LKV semantics do not fit.
• Sources whose payloads exceed MQTT broker per-message limits.
• Sources that need consumer-side replay/seek to historical offsets. MQTT retained slots only hold the last value per topic.

Mandatory Expert Reviews

Three subagents must review the MQTT design before generation runs:

Aspect	Reviewer	What to ask
Topic tree	UNS Catalog Architect if available; otherwise consult the xRegistry Expert who knows the UNS-on-xRegistry mapping.	Validate the hierarchy depth, segment naming, wildcard fitness, retained vs non-retained per event type, QoS, single-vs-multi-topic-per-object decisions.
xRegistry MQTT contract	xRegistry Expert	Validate the MQTT/5.0 endpoint, the dedicated <source>.mqtt messagegroup with basemessageurl references back into the transport-neutral group, and the protocoloptions.properties (topic / qos / retain).
Schemas	JSON Structure Expert	Re-confirm that JsonStructure schemas (shared with Kafka) still cover every field; add no MQTT-specific schema drift.

Do not skip these reviews. Even when the same human contract author also wrote the Kafka contract, MQTT-specific decisions (retained slots, wildcards, key/topic alignment) require an independent expert pass.

Mandatory Pre-merge Documentation Gate

A feeder PR is NOT mergeable until every one of the following docs mentions MQTT in a meaningful way. Both reviewers (xRegistry + UNS) and the implementing agent must verify this BEFORE opening the PR. Reject or fix any PR that fails the gate.

File	Required MQTT content
<source>/README.md	A "Transports" (or equivalently named) section that names the MQTT app, explains the Kafka vs MQTT choice for this source, and lists all four deployment templates (Kafka container, Kafka + Event Hubs, MQTT container, MQTT + Event Grid namespace).
<source>/CONTAINER.md	An MQTT section with env-var table (broker host, port, TLS, auth mode, topic prefix, retain default, QoS default) and the two MQTT Deploy-to-Azure badges.
<source>/EVENTS.md	The MQTT messagegroup with the literal topic patterns (e.g. air-quality/be/issep/...) per message type. Must be regenerated from the updated xreg, not hand-edited.
Root README.md	The source's row appends the MQTT image link and the two MQTT deploy badges.
ghpages/app.js and ghpages/catalog.json	mqtt: true set on the source entry so the portal renders the MQTT deploy buttons.

Each of these MUST be touched in the same PR as the MQTT feeder code. PRs that ship code without docs are bug-causing and have happened repeatedly — do not allow it.

Reviewer checklist (paste into PR description or review comment):

• <source>/README.md mentions MQTT and the four deploy templates
• <source>/CONTAINER.md has MQTT env-var table + two MQTT badges
• <source>/EVENTS.md shows the MQTT messagegroup + topic patterns
• Root README.md row has MQTT image + two MQTT badges
• ghpages/app.js and ghpages/catalog.json set mqtt: true

If any box is unchecked, the PR is incomplete — either fix in branch or re-open as a "doc backfill" PR before declaring the feeder done.

Mandatory AMQP Companion

Every MQTT feeder PR in this repo MUST also ship an AMQP 1.0 sibling feeder in the same PR. This is non-negotiable. The three transports (Kafka, MQTT, AMQP) form the standard trio for every source.

Rationale: MQTT is right for UNS-style consumers (low-latency, retained LKV, pub/sub fan-out). AMQP 1.0 is right for queue-oriented enterprise consumers (Azure Service Bus, ActiveMQ Artemis, Qpid Dispatch, RabbitMQ AMQP 1.0 plugin) where work-distribution semantics or transactional delivery matter. Shipping one without the other leaves half the consumer population unserved and creates ongoing per-source retrofit debt — which we are no longer willing to take on.

Authoritative contract: .github/skills/amqp-feeder/SKILL.md. Read it in full alongside this document. Pegelonline is the reference implementation for both transports.

Required AMQP artifact	Where
<source>.amqp messagegroup + AMQP endpoint(s) in xreg	<source>/xreg/<source>.xreg.json
AMQP CBS target generated via xrcg --azure_cbs_target	<source>_amqp_producer/... (generated)
<source>_amqp/ app folder sharing the core acquisition module	sibling of <source>_kafka/ and <source>_mqtt/
Dockerfile.amqp	<source>/Dockerfile.amqp
Test<Source>AmqpDockerFlow class	tests/docker_e2e/test_docker_amqp_flow.py
AMQP rows in matrix.json (build + flow)	tests/docker_e2e/matrix.json
ARM template for Service Bus (CBS / Entra ID)	<source>/azure-template-amqp.json
AMQP section in CONTAINER.md with env-var table + Deploy badges	<source>/CONTAINER.md
AMQP messagegroup + address templates in EVENTS.md	<source>/EVENTS.md
Root README.md row: AMQP image link + AMQP deploy badges	README.md
amqp: true on the portal entry	ghpages/app.js + ghpages/catalog.json
<source>/README.md Transports section lists Kafka + MQTT + AMQP	<source>/README.md

The AMQP messagegroup MUST share the schemas (single schemagroups.<source>.jstruct block) and use the SAME identity tuple as the Kafka subject/key and the MQTT topic. AMQP address templates match the topic placeholders one-for-one.

Reviewer checklist (paste into PR description or review comment):

• xreg has <source>.amqp messagegroup with AMQP endpoint(s) (generic SASL PLAIN + Azure Service Bus CBS)
• xrcg regenerated with --azure_cbs_target and the generated <source>_amqp_producer is committed
• <source>_amqp/ app folder exists, shares core acquisition module with Kafka/MQTT apps
• Dockerfile.amqp builds; OCI labels set correctly
• Test<Source>AmqpDockerFlow class exists in test_docker_amqp_flow.py
• matrix.json has the AMQP build row AND the AMQP flow row
• CI shows the AMQP flow job actually executing and passing
• azure-template-amqp.json present and referenced from CONTAINER.md
• CONTAINER.md, EVENTS.md, source README, root README, portal entries all mention AMQP
• AMQP identity placeholders match Kafka subject/key and MQTT topic exactly

A PR that adds MQTT without AMQP is not mergeable — open it as Kafka+MQTT+AMQP from day one. If you discover mid-implementation that AMQP is impossible for this source (e.g. upstream payloads cannot be addressed under any sensible AMQP address template), document the reason in the PR body and request explicit user approval to ship MQTT-only as an exception.

Mandatory Pre-merge Test Gate

A feeder PR is NOT mergeable until both of the following exist in the same PR as the feeder code. This has been violated repeatedly; do not allow it.

Required	Where
Test<Source>MqttDockerFlow class	tests/docker_e2e/test_docker_mqtt_flow.py
Matrix build entries (one per image: <source> Kafka and <source>-mqtt)	tests/docker_e2e/matrix.json build array
Matrix flow entry with "test_file": "test_docker_mqtt_flow.py"	tests/docker_e2e/matrix.json flow array

The Docker E2E test must boot eclipse-mosquitto:2 alongside the <source>_mqtt image and assert everything described in the Docker E2E section below (retained refs, telemetry, topic match, binary-mode CE properties, schema validation).

Reviewer checklist (paste into PR description or review comment):

• Test<Source>MqttDockerFlow class exists in test_docker_mqtt_flow.py
• matrix.json build has the <source>-mqtt image entry
• matrix.json flow has a row with "test_file": "test_docker_mqtt_flow.py" pointing at the new class
• CI run on the PR shows the new MQTT flow job actually executing and passing (not skipped by path filter)

A "fast-shipping" feeder PR that omits the test is a regression, not a feature. Reject it.

Inputs

• source id (e.g. pegelonline)
• list of event types and their identity tuple (e.g. Station keyed on water_shortname/station_id, CurrentMeasurement keyed on the same)
• decision: which events are retained (LKV) and which are not
• the UNS domain prefix the source belongs to (e.g. hydro/de/wsv/pegelonline/...)
• whether a Kafka feeder already exists in the folder

Non-Negotiables

• One process, one transport. Never emit MQTT and Kafka from the same Python process or container.
• The MQTT and K

---

Content truncated.