mqtt-uns-feeder

name	mqtt-uns-feeder
description	Use when adding MQTT/Unified-Namespace transport to a feeder in this repo — either alongside an existing Kafka feeder (transport split) or for a brand-new source where MQTT is the primary fit. Covers UNS topic-tree design, xRegistry MQTT messagegroup pattern, paho-mqtt v5 binary-mode CloudEvent emission, Entra JWT enhanced auth for Event Grid namespace brokers, two-app folder layout sharing a core acquisition module, Dockerfile.mqtt + dedicated pyproject, two MQTT ARM templates (BYO broker + Event Grid namespace), the read-only test-client registration helper, MQTT Docker E2E, and the four-button portal/README update. NOTE: AMQP 1.0 transport is MANDATORY alongside MQTT — see `.github/skills/amqp-feeder/SKILL.md` and the 'Mandatory AMQP Companion' section below.
argument-hint	Describe the source id, what data families it emits, the upstream identifiers that will form the UNS topic hierarchy, and whether the source already has a Kafka feeder (transport split) or this is greenfield MQTT-only.

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 Kafka messagegroups share schemas (single schemagroups.<source>.jstruct block). MQTT messages declare basemessageurl back to the transport-neutral message definitions.
• The MQTT topic uritemplate must use the same identity tuple as the Kafka subject and key. The placeholder names match across all three.
• All MQTT payloads use binary-mode CloudEvents (CloudEvents metadata in MQTT v5 user properties / system properties; payload is the raw JSON body). Never use structured-mode JSON envelopes.
• Reference events (catalog data) are emitted with retain: true once at startup and re-emitted on periodic refresh.
• Telemetry events whose latest value is meaningful in isolation are also retained (LKV). Telemetry that is only meaningful as a stream (e.g. discrete alerts, event streams) is not retained.
• The MQTT app is generated through xrcg 0.10.8 exactly as Kafka is. The generated producer is never hand-edited.
• Authentication is configurable at runtime (anonymous / username-password / TLS-cert / Entra JWT). The contract does not pin the auth mode.

Recommended UNS Topic-Tree Shape

<domain>/<country>/<authority>/<source>/<sub-group>/<object-id>/<event-type>

• <domain> is the broad data domain (hydro, air, traffic, maritime, energy, seismic, weather, ...).
• <country> is the ISO 3166-1 alpha-2 code (lower-case) where the upstream authority operates.
• <authority> is the issuing body (wsv, noaa, dwd, bom).
• <source> is the canonical source id (matches the repo folder name).
• <sub-group> is a stable upstream grouping that aids wildcard subscriptions (basin, region, agency, vessel-class). Skip this level only if no such grouping exists.
• <object-id> is the stable per-object identifier (station UUID, MMSI, gauge number, sensor id).
• <event-type> is a kebab-cased noun describing what the topic carries (info, water-level, metar, alert, ...).

Rules:

• Lower-case kebab-case everywhere. No human names. No display labels.
• Use the same identity placeholders as the Kafka key/subject template (verbatim placeholder names).
• One retained, QoS-1 topic per (object, event-type) pair. Subscribers joining the broker mid-stream see the latest state immediately.
• Subscribers consuming a sub-tree wildcard (e.g. hydro/de/wsv/+/+/+/water-level) must get a coherent slice of state without needing replay.

Worked example (pegelonline):

hydro/de/wsv/pegelonline/<water_shortname>/<station_id>/info          (retained, QoS-1)
hydro/de/wsv/pegelonline/<water_shortname>/<station_id>/water-level   (retained, QoS-1)

xRegistry Contract Pattern

Add two siblings to the existing transport-neutral content:

1. A new endpoint <rev-dns>.<Source>.Mqtt with protocol: "MQTT/5.0" that points at #/messagegroups/<rev-dns>.<source>.mqtt.
2. A new messagegroup <rev-dns>.<source>.mqtt whose messages each carry:
   • basemessageurl referencing the transport-neutral message in <rev-dns>.<source> (so the schema is shared);
   • protocol: "MQTT/5.0";
   • protocoloptions.properties.topic of type uritemplate with a value matching the Kafka subject/key shape;
   • protocoloptions.properties.qos (integer, typically 1);
   • protocoloptions.properties.retain (boolean, per event-type retention decision).

Look at pegelonline/xreg/pegelonline.xreg.json lines 23–41 (endpoint) and 119–167 (messagegroup) for the canonical shape.

Do not create a second schemagroups block. The MQTT messagegroup references the same JsonStructure schemas the Kafka messagegroup uses via basemessageurl.

After editing the manifest:

./generate_producer.ps1   # already exists; regenerates both Kafka and MQTT

xrcg 0.10.8 produces a generated MQTT producer sub-package alongside the Kafka one. Treat it as installable but never editable.

Repository Folder Layout (transport split)

<source>/
  README.md
  CONTAINER.md
  EVENTS.md
  Dockerfile.kafka                 # CMD: python -m <source>_kafka feed
  Dockerfile.mqtt                  # CMD: python -m <source>_mqtt  feed
  generate_producer.ps1
  xreg/<source>.xreg.json
  <source>/                        # shared core acquisition + config
    __init__.py
    acquisition.py
    config.py
    state.py
  <source>_kafka/
    pyproject.toml                 # depends on the generated Kafka producer
    <source>_kafka/
      __init__.py
      __main__.py
      app.py                       # owns the Kafka producer lifecycle
  <source>_mqtt/
    pyproject.toml                 # depends on the generated MQTT producer
    <source>_mqtt/
      __init__.py
      __main__.py
      app.py                       # owns the MQTT producer lifecycle
  <source>_producer/               # xrcg output — never edited
  azure-template.json              # Kafka, BYO Event Hub
  azure-template-with-eventhub.json
  azure-template-mqtt.json         # MQTT, BYO broker
  azure-template-with-eventgrid-mqtt.json
  tests/
    test_<source>_kafka_app.py
    test_<source>_mqtt_app.py

The shared core package (<source>/) owns everything that is identical between transports: HTTP/MQTT/WS acquisition of the upstream, config parsing, state/dedupe, normalization into the generated data classes. Both <source>_kafka/app.py and <source>_mqtt/app.py import from it.

Runtime Bridge Pattern (MQTT app)

Use paho-mqtt >= 2.1 as the only MQTT client. The generated producer already wraps it.

• Parse source + MQTT configuration from env vars and CLI args (see MQTT environment variables).
• Pick the auth mode at runtime: anonymous | userpass | tls-cert | entra.
• Connect (see auth-specific notes below), then start the paho loop.
• Emit reference events first (catalog) as retained QoS-1, then enter the telemetry loop.
• Re-emit reference events on a periodic refresh (same cadence as the Kafka feeder).
• Pass topic placeholders explicitly to the generated producer's send_* methods — never hide them inside a key/topic mapper.

MQTT environment variables

The contract:

Variable	Required	Default	Description
MQTT_BROKER_URL	yes	—	host:port of the broker. TLS implied at 8883 / configurable.
MQTT_ENABLE_TLS	no	true	Enable TLS. Always true for cloud brokers.
MQTT_AUTH_MODE	no	anonymous	anonymous / userpass / tls-cert / entra.
MQTT_USERNAME	conditional	—	For userpass. Also used as MQTT username in entra (the Client resource name on Event Grid).
MQTT_PASSWORD	conditional	—	For userpass.
MQTT_CLIENT_CERT	conditional	—	PEM path for tls-cert.
MQTT_CLIENT_KEY	conditional	—	PEM path for tls-cert.
MQTT_CA_FILE	no	system trust	Path to the broker CA chain.
MQTT_CLIENT_ID	no	source-derived	MQTT client identifier.
MQTT_ENTRA_CLIENT_ID	conditional	—	User-assigned MI client id (used as ManagedIdentityCredential(client_id=…)).
MQTT_ENTRA_AUDIENCE	no	https://eventgrid.azure.net/	JWT audience. Must include the trailing slash for EG namespace MQTT.

Always make MQTT_BROKER_URL the single point where the broker hostname is configured. Never hard-code it.

Auth: Entra JWT enhanced authentication (Event Grid namespace)

This is the only complicated path. Steps:

1. Acquire a token via azure.identity.ManagedIdentityCredential(client_id=MQTT_ENTRA_CLIENT_ID) (or DefaultAzureCredential as fallback) for resource MQTT_ENTRA_AUDIENCE.
2. Build paho.mqtt.properties.Properties(PacketTypes.CONNECT) with:
   • AuthenticationMethod = "OAUTH2-JWT"
   • AuthenticationData = <jwt bytes>
3. The generated MqttClient.connect(broker, port, keepalive) signature does not accept properties, so bypass it: call paho_client.connect(host, port, keepalive=60, clean_start=True, properties=props) then paho_client.loop_start() directly. Take the underlying paho client off the generated wrapper's .client attribute.
4. Start a background asyncio task that refreshes the token ~5 minutes before expiry. paho-mqtt 2.1 has no reauth(), so refresh = call paho_client.disconnect() and reconnect with new CONNECT properties.
5. Apply Azure RBAC at topic-space scope, role EventGrid TopicSpaces Publisher (a12b0b94-b317-4dcd-84a8-502ce99884c6). No clients / clientGroups / permissionBindings are needed when using RBAC.

pegelonline/pegelonline_mqtt/pegelonline_mqtt/app.py is the reference implementation — copy its _acquire_entra_token helper and the _entra_token_refresh_loop shape verbatim.

Auth: X.509 client certificate

Used by the legacy clients model (and by the testing tool):

• Configure paho via tls_set(ca_certs=..., certfile=..., keyfile=...).
• The broker matches a Client resource by thumbprint or subject CN.
• Authorization is granted via permission bindings on ClientGroups selected by attributes.<name> = '<value>' queries.

pyproject.toml — MQTT sibling

[build-system]
requires = ["poetry-core>=1.1.0"]
build-backend = "poetry.core.masonry.api"

[tool.poetry]
name = "<source>-mqtt"
version = "0.1.0"
packages = [{include = "<source>_mqtt"}]

[tool.poetry.dependencies]
python = ">=3.10,<4.0"
requests = ">=2.32.3"
paho-mqtt = ">=2.1.0"
cloudevents = ">=1.11.0,<2.0.0"
dataclasses_json = ">=0.6.7"
azure-identity = ">=1.17.0"     # only when MQTT_AUTH_MODE=entra is supported
<source>_producer_data         = {path = "../<source>_producer/<source>_producer_data"}
<source>_producer_mqtt_producer = {path = "../<source>_producer/<source>_producer_mqtt_producer"}

Install order in containers: producer_data → producer_mqtt_producer → <source>-mqtt.

Dockerfile.mqtt convention

FROM python:3.10-slim
LABEL source="<source>" \
      title="<Source> MQTT feeder" \
      description="<one-liner>" \
      documentation="https://github.com/clemensv/real-time-sources/blob/main/<source>/CONTAINER.md" \
      license="MIT"
# … install order: producer_data → producer_mqtt_producer → <source>_mqtt
# Build-time import smoke: a successful build proves <source>_mqtt and its
# companion packages import inside the image (closes the "built but never
# ran" gap). PR-time equivalent: tools/ci/import_smoke.py.
RUN python -c "import importlib, importlib.util as _u; _m = '<source>_mqtt'; importlib.import_module(_m + '.app' if _u.find_spec(_m + '.app') else _m)"
CMD ["python", "-m", "<source>_mqtt", "feed"]

GHCR image name convention: ghcr.io/clemensv/real-time-sources-<source>-mqtt:latest (distinct from the Kafka image which is …-<source>-kafka:latest).

Azure Deployment Templates

Two MQTT-specific templates must ship alongside the existing two Kafka ones:

azure-template-mqtt.json — BYO MQTT broker

• Parameters: mqttBrokerUrl, optional mqttUsername/mqttPassword, imageName, image registry creds.
• Provisions: an ACI running the …-<source>-mqtt:latest image.
• Env: MQTT_BROKER_URL, MQTT_AUTH_MODE (defaults to userpass or anonymous based on inputs).

azure-template-with-eventgrid-mqtt.json — full Event Grid namespace

Provisions, in order:

1. Microsoft.EventGrid/namespaces with topicSpacesConfiguration.state = 'Enabled'.
2. A topicSpaces child with template <topicRoot> (default <domain>/#).
3. A user-assigned managed identity.
4. A role assignment of EventGrid TopicSpaces Publisher (a12b0b94-b317-4dcd-84a8-502ce99884c6) scoped to the topic space.
5. An ACI with the user-assigned MI attached and env vars:
   • MQTT_AUTH_MODE=entra
   • MQTT_ENTRA_CLIENT_ID = <MI.clientId>
   • MQTT_BROKER_URL = reference(<ns>, '2025-07-15-preview').topicSpacesConfiguration.hostname:8883

ARM gotchas (do not repeat these mistakes)

• The role assignment is scoped to the topic space, not the resource group. In the ACI dependsOn, reference it via extensionResourceId(<topicSpaceResourceId>, 'Microsoft.Authorization/roleAssignments', <name>). Plain resourceId('Microsoft.Authorization/roleAssignments', name) resolves to the RG scope and the template will fail validation.
• The broker hostname must be live-sourced with reference(<namespaceResourceId>, '2025-07-15-preview').topicSpacesConfiguration.hostname. Do not assemble it from the region — the Event Grid team has reserved the right to change the suffix.
• The container image defaults to the public ghcr.io image published by the repo build; no imageRegistryCredentials are needed. Add them back only if you fork and host the image in a private registry.

Use pegelonline/azure-template-with-eventgrid-mqtt.json as the template — copy and adapt the namePrefix / topicRoot / imageName.

Testing Helper

tools/Register-EventGridMqttTestClient.ps1 registers a read-only client on any Event Grid namespace, generates a self-signed cert, fetches the broker's CA chain live, and drops .crt / .key / -ca.crt files into the current directory. Use it for MQTTX inspection during development. The script grants only the Subscriber permission on the chosen topic space (publishing is denied).

Docker E2E

Ship a sibling test class in tests/docker_e2e/test_docker_mqtt_flow.py that:

1. Boots eclipse-mosquitto:2 alongside the <source>_mqtt image in the same compose-style topology used by the Kafka E2E.
2. Subscribes a paho client to the source's UNS root wildcard.
3. Asserts:
   • At least one retained message for every reference topic.
   • At least one telemetry message under each declared event type.
   • Topic strings match the uritemplate from the xreg manifest.
   • Binary-mode CloudEvent properties (ce_specversion, ce_type, ce_source, ce_subject, ce_id, ce_time, content_type) are present in the MQTT v5 user properties.
   • Payload validates against the JsonStructure schema in the manifest.

Reuse the settled compose networking from the Kafka/MQTT E2E harness verbatim -- the broker hostname, the shared Docker network, and the container-to-container address resolution were stabilized over many trial-and-error iterations (and one revert). Copy the existing test_docker_mqtt_flow.py topology; do not re-derive the networking.

A source is not done until this passes and at least one real MQTT message has been observed on the wire for the source -- a green build, a started container, and passing unit tests do not count as proof (the class-A "shipped but never ran" defects all passed those and emitted zero messages). Record the observed-message evidence in the PR body.

Documentation Surface

• CONTAINER.md — split env-var tables (Kafka section + MQTT section), four Deploy-to-Azure badges (Kafka container, Kafka container + Event Hubs, MQTT container, MQTT container + Event Grid namespace), MS-Learn link to the EG namespace MQTT broker docs.
• EVENTS.md — regenerate so it includes the MQTT messagegroup with topic patterns.
• README.md — add a "Transports" section that explains the Kafka-vs-MQTT choice and lists the four deployment templates.
• Root README.md — append the MQTT image and the two MQTT deploy badges to the source's row.
• Portal hooks — set mqtt: true on the source entry in both app.js and catalog.json on main. Two extra buttons (btn-container-mqtt, btn-container-mqtt-eg) already exist in the ghpages branch's index.html; they auto-show for any source whose catalog entry carries mqtt: true. Two hash deep-links also work: #<id>/azure-mqtt and #<id>/azure-mqtt-eg.

Implementation Sequence

Three transports, one PR. Every step that touches MQTT below has a mandatory AMQP twin in the same PR (see "Mandatory AMQP Companion" above and .github/skills/amqp-feeder/SKILL.md). Do not interleave a separate "AMQP follow-up" PR — that has caused retrofit debt repeatedly and is no longer allowed.

1. Topic-tree + address-template review with the UNS expert (or the xRegistry Expert if the UNS agent is not available). Both MQTT topic templates AND AMQP address templates must be designed in this step using the SAME identity tuple.
2. xRegistry edit — add the Mqtt endpoint + the <source>.mqtt messagegroup AND the AMQP endpoint(s) + the <source>.amqp messagegroup. Run xRegistry Expert review.
3. Regenerate producers via ./generate_producer.ps1 (covers Kafka, MQTT, AND AMQP — the script must invoke xrcg three times, including the AMQP run with --azure_cbs_target).
4. Folder split — <source>_kafka/, <source>_mqtt/, <source>_amqp/ siblings all sharing the core acquisition module.
5. MQTT runtime bridge in <source>_mqtt/<source>_mqtt/app.py AND AMQP runtime bridge in <source>_amqp/<source>_amqp/app.py, both sharing the core acquisition module with the Kafka app.
6. Unit + integration tests for both MQTT and AMQP apps under tests/test_<source>_mqtt_app.py and tests/test_<source>_amqp_app.py.
7. Dockerfile.mqtt and Dockerfile.amqp with their pyprojects.
8. ARM templates — two MQTT (BYO broker + Event Grid namespace) AND one AMQP (Service Bus / generic).
9. Docker E2E — MQTT class in test_docker_mqtt_flow.py AND AMQP class in test_docker_amqp_flow.py, with matching matrix.json build + flow rows for both transports.
10. Doc refresh: CONTAINER.md (Kafka/MQTT/AMQP sections), EVENTS.md (all three messagegroups), README.md (Transports section lists six deployment templates), root README.md (MQTT + AMQP badges).
11. Portal hooks: mqtt: true AND amqp: true in app.js + catalog.json.
12. Live smoke test — recommended for MQTT (Register-EventGridMqttTestClient.ps1 + MQTTX) AND for AMQP (local Artemis or Service Bus + a CLI AMQP consumer).

Fleet Dispatch Prompt Template

When dispatching a batch agent to add new feeders or retrofit existing ones, the prompt MUST include the following non-negotiables:

1. Read this skill (.github/skills/mqtt-uns-feeder/SKILL.md) AND .github/skills/amqp-feeder/SKILL.md in full before starting.
2. Every PR ships Kafka + MQTT + AMQP together. PRs that omit AMQP are rejected and force a rework loop — do not open them.
3. Walk the three pre-merge gates against your own work before opening the PR:
   • Mandatory Pre-merge Documentation Gate
   • Mandatory AMQP Companion
   • Mandatory Pre-merge Test Gate
4. Dispatch the mandatory expert reviews (xRegistry, JSON Structure, Avro Schema, UNS Catalog Architect; CloudEvents + Event Streaming recommended for AMQP) and address every blocker before merging.
5. Walk the container-and-delivery and feeder-release-checklist skills end-to-end. Record verdicts in the PR body.
6. Verify CI proof-of-execution: the MQTT flow job AND the AMQP flow job actually executed and passed (not just queued, not skipped by path filter).

If any of these is missing from the dispatch prompt, refuse to start work and ask the dispatcher to re-issue with the required scope.

Common Failure Modes

Symptom	Cause	Fix
Connect succeeds, no events flow	Forgot to override the generated MqttClient.connect() for entra auth	Bypass the generated wrapper; call paho.connect(..., properties=connect_props) directly.
Connect rejected with NOT_AUTHORIZED	Wrong RBAC scope or missing role assignment	Assign EventGrid TopicSpaces Publisher (a12b0b94-…) at the topic-space scope, not the namespace or RG.
Connect rejected after ~24 h	Entra token expired, no refresh loop	Add the disconnect+reconnect refresh task. paho 2.1 has no reauth().
Subscriber sees no retained reference data	Telemetry was emitted before reference data, or retain: true was missing in the contract	Emit reference data first; verify retain: true is present in protocoloptions.properties.retain.
Template validation error on the role assignment	dependsOn uses resourceId('Microsoft.Authorization/roleAssignments', name)	Use extensionResourceId(<topicSpaceResourceId>, 'Microsoft.Authorization/roleAssignments', name).
Hostname looks wrong (region suffix changed)	Hardcoded <ns>.<region>-1.ts.eventgrid.azure.net	Use reference(<ns>, '2025-07-15-preview').topicSpacesConfiguration.hostname.
ConvertFrom-Json fails in tooling	az CLI emits SyntaxWarnings to stderr that get mixed into stdout	Capture stdout only (no 2>&1); let stderr pass through.
Test client cannot validate broker cert	Self-signed client cert was used as the "CA file"	Fetch the broker's server-cert chain at runtime (see Register-EventGridMqttTestClient.ps1).
MQTTX shows "publish failed" but no error code	Broker is silently denying the publish	The client only holds Subscriber permission; this is correct. Add a Publisher binding for a different identity to test publishing.

References

• Reference implementation: pegelonline/pegelonline_mqtt/pegelonline_mqtt/app.py
• Reference xreg: pegelonline/xreg/pegelonline.xreg.json
• Reference ARM (EG namespace): pegelonline/azure-template-with-eventgrid-mqtt.json
• Reference Dockerfile: pegelonline/Dockerfile.mqtt
• Reference Docker E2E: tests/docker_e2e/test_docker_mqtt_flow.py
• Testing helper: tools/Register-EventGridMqttTestClient.ps1
• xrcg generator: https://github.com/clemensv/xrcg (pinned at 0.10.1)
• Event Grid namespace MQTT broker docs: https://learn.microsoft.com/azure/event-grid/mqtt-overview
• MQTT v5 Enhanced Authentication for OAuth/JWT (Event Grid namespace): https://learn.microsoft.com/azure/event-grid/mqtt-client-microsoft-entra-token-authentication