MQTT 5 Request/Response mit Response Topic und Correlation Data testen

MQTT ist Publish/Subscribe, aber viele reale Systeme brauchen Frage und Antwort: „Lies diese Einstellung“, „führe diesen Befehl aus und sag mir, ob es geklappt hat“. MQTT 5 Request/Response standardisiert das mit zwei PUBLISH-Properties: Der Anfragende setzt ein Response Topic (wohin die Antwort gehen soll) und Correlation Data (eine opake ID, die Antwort und Frage verknüpft). Der Responder kopiert die Correlation Data in seine Antwort und publisht auf das Response Topic.

Diese Anleitung implementiert das Muster in Python und JavaScript, zeigt einen Einzeiler-Test auf der Kommandozeile und listet die Testfälle auf, die genau die Bugs finden, die sonst in Produktion landen: nicht passende IDs, fehlende Timeouts, parallele Anfragen und ACLs, die die Antwort blockieren. Für die Design-Seite des Musters siehe MQTT Request/Response.

So funktioniert der Ablauf

Der Broker erzwingt hier nichts – er leitet die Properties einfach weiter. Gut für die Portabilität (jeder MQTT-5-Broker funktioniert), schlecht für die Korrektheit: Vergisst dein Responder, die Correlation Data zu kopieren, beschwert sich niemand, bis der Anfragende in den Timeout läuft.

Es gibt zwei verwandte CONNECT/CONNACK-Properties. Ein Client kann im CONNECT Request Response Information setzen, und der Broker kann im CONNACK mit Response Information antworten – einem Topic-Präfix, das der Client für Antworten verwenden soll. Die Unterstützung ist je nach Broker unterschiedlich, deshalb bauen die meisten Anwendungen das Response Topic einfach aus ihrer Client-ID. So oder so ist das Response Topic ein Topic-Name: Es darf weder + noch # enthalten.

Zuerst auf der Kommandozeile testen

Mosquitto liefert mosquitto_rr mit, einen Request/Response-Client, der standardmäßig MQTT 5 spricht. Starte in einem Terminal einen Fake-Responder – einen Subscriber, der die Properties ausgibt –, damit du siehst, was ankommt:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -t 'testmqtt/rr/req/#' \
  -F 'topic=%t response=%R corr=%D payload=%p'

Sende in einem zweiten Terminal eine Anfrage mit mosquitto_rr. Es abonniert das Response Topic, publisht die Anfrage und wartet auf eine Antwort:

mosquitto_rr -h broker.connectmq.com -p 1883 \
  -t testmqtt/rr/req/device-7 \
  -e testmqtt/rr/resp/cli-$$ \
  -m '{"op":"get","key":"firmware"}' -W 5

Terminal 1 zeigt Response Topic und Correlation Data. Um von Hand zu antworten, publishe aus einem dritten Terminal auf das angezeigte Response Topic mit derselben Correlation Data:

mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -t testmqtt/rr/resp/cli-12345 \
  -D publish correlation-data '<in Terminal 1 ausgegebener Wert>' \
  -m '{"firmware":"2.4.1"}'

mosquitto_rr gibt die Antwort aus und beendet sich; mit -W 5 gibt es nach fünf Sekunden auf – das ist dein erster Timeout-Test. Du kannst beide Topics auch im Online-MQTT-Client beobachten: Abonniere testmqtt/rr/#, und jede Anfrage und Antwort erscheint nebeneinander.

Responder und Anfragender in Python

Das Folgende nutzt paho-mqtt 2.x mit MQTT 5. Der Responder abonniert die Anfragen und schickt die Correlation Data zurück:

import json
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

HOST = "broker.connectmq.com"

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe("testmqtt/rr/req/+", qos=1)

def on_message(client, userdata, msg):
    resp_topic = getattr(msg.properties, "ResponseTopic", None)
    corr = getattr(msg.properties, "CorrelationData", None)
    if not resp_topic:
        print("Anfrage ohne Response Topic, ignoriert:", msg.topic)
        return
    request = json.loads(msg.payload)
    reply = {"ok": True, "echo": request}
    props = Properties(PacketTypes.PUBLISH)
    if corr is not None:
        props.CorrelationData = corr
    client.publish(resp_topic, json.dumps(reply), qos=1, properties=props)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_message = on_message
client.connect(HOST, 1883)
client.loop_forever()

Der Anfragende führt eine Tabelle offener Anfragen, indiziert nach Correlation-ID, sodass viele Anfragen gleichzeitig unterwegs sein und Antworten in beliebiger Reihenfolge eintreffen können:

import json, threading, time, uuid
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

HOST = "broker.connectmq.com"
CLIENT_ID = f"rr-test-{uuid.uuid4().hex[:8]}"
RESP_TOPIC = f"testmqtt/rr/resp/{CLIENT_ID}"
pending = {}  # Correlation-ID -> [Event, Antwort]
ready = threading.Event()

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(RESP_TOPIC, qos=1)

def on_subscribe(client, userdata, mid, reason_codes, properties):
    ready.set()

def on_message(client, userdata, msg):
    corr = getattr(msg.properties, "CorrelationData", b"")
    slot = pending.get(corr)
    if slot is None:
        print("unerwartete oder verspätete Antwort:", corr)
        return
    slot[1] = json.loads(msg.payload)
    slot[0].set()

def request(client, device, body, timeout=5.0):
    corr = uuid.uuid4().bytes
    slot = pending[corr] = [threading.Event(), None]
    props = Properties(PacketTypes.PUBLISH)
    props.ResponseTopic = RESP_TOPIC
    props.CorrelationData = corr
    props.MessageExpiryInterval = int(timeout)  # veraltete Anfragen werden verworfen
    t0 = time.perf_counter()
    client.publish(f"testmqtt/rr/req/{device}", json.dumps(body), qos=1, properties=props)
    ok = slot[0].wait(timeout)
    pending.pop(corr, None)
    if not ok:
        raise TimeoutError(f"keine Antwort von {device} innerhalb von {timeout}s")
    return slot[1], (time.perf_counter() - t0) * 1000

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID, protocol=mqtt.MQTTv5)
client.on_connect, client.on_subscribe, client.on_message = on_connect, on_subscribe, on_message
client.connect(HOST, 1883)
client.loop_start()
ready.wait(5)

reply, ms = request(client, "device-7", {"op": "get", "key": "firmware"})
print(f"Antwort in {ms:.1f} ms: {reply}")

Zwei Details sind entscheidend. Der Anfragende wartet vor dem Senden auf das SUBACK – sonst kann ein schneller Responder antworten, bevor das Abonnement existiert, und die Antwort geht verloren. Und die Anfrage trägt ein Message Expiry Interval in Höhe des Timeouts, damit ein Responder, der später online kommt, keinen Befehl ausführt, den der Aufrufer schon aufgegeben hat (siehe Message Expiry).

Dasselbe Muster mit MQTT.js

MQTT.js 5 stellt die Properties in den Publish-Optionen und am empfangenen Paket bereit:

import mqtt from "mqtt";
import { randomUUID } from "node:crypto";

const id = `rr-js-${randomUUID().slice(0, 8)}`;
const respTopic = `testmqtt/rr/resp/${id}`;
const client = await mqtt.connectAsync("mqtt://broker.connectmq.com:1883", {
  protocolVersion: 5,
  clientId: id,
});
await client.subscribeAsync(respTopic, { qos: 1 });

const pending = new Map();
client.on("message", (topic, payload, packet) => {
  const corr = packet.properties?.correlationData?.toString();
  pending.get(corr)?.(JSON.parse(payload.toString()));
  pending.delete(corr);
});

function request(device, body, timeoutMs = 5000) {
  const corr = randomUUID();
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      pending.delete(corr);
      reject(new Error(`Timeout beim Warten auf ${device}`));
    }, timeoutMs);
    pending.set(corr, (reply) => { clearTimeout(timer); resolve(reply); });
    client.publish(`testmqtt/rr/req/${device}`, JSON.stringify(body), {
      qos: 1,
      properties: { responseTopic: respTopic, correlationData: Buffer.from(corr) },
    });
  });
}

console.log(await request("device-7", { op: "get", key: "firmware" }));
await client.endAsync();

Lass es gegen den Python-Responder von oben laufen: Beide Libraries arbeiten zusammen, weil sie nur Standard-Properties von MQTT 5 verwenden. Der Code-Generator erzeugt den Verbindungsteil für weitere Sprachen.

Testfälle, die sich zu automatisieren lohnen

Eine Request/Response-Schicht wirkt trivial, bis sie in Produktion versagt. Diese Fälle decken die meisten echten Bugs ab:

1. Happy Path. Eine Anfrage, eine Antwort; Payload und Correlation Data stimmen überein.
2. Responder offline. Der Anfragende muss einen Timeout auslösen, nicht hängen bleiben. Miss, dass er beim konfigurierten Wert feuert.
3. Verspätete Antwort. Lass den Responder länger schlafen als der Timeout. Die späte Antwort muss ignoriert und darf nicht der nächsten Anfrage zugestellt werden.
4. Parallelität. Feuere 100 Anfragen parallel ab, mit einem Responder, der zufällige Verzögerung einbaut; jede Antwort muss beim richtigen Aufrufer landen. Antworten außer der Reihe sind normal.
5. Fehlende Properties. Sende eine Anfrage ohne Response Topic. Der Responder soll sie loggen und verwerfen, nicht abstürzen.
6. Abgelaufene Anfrage. Stoppe den Responder, sende eine Anfrage mit 2 Sekunden Expiry, warte 5 Sekunden und starte den Responder mit persistenter Session neu. Er darf die Anfrage nicht ausführen.
7. Autorisierung. Mit aktiven ACLs braucht der Responder Publish-Rechte auf die Response Topics und der Anfragende Subscribe-Rechte auf sein eigenes. Eine verweigerte Antwort wird oft ohne Fehler verworfen – siehe ACL still verweigert.

Wenn die Antwort nie ankommt

Eine nützliche Gewohnheit: Lass den Responder das empfangene Response Topic und die Correlation Data vor der Verarbeitung loggen. Schlägt ein Test fehl, sagt dir diese eine Logzeile, welche Seite falsch liegt.

Round-Trip-Latenz messen

Da der Anfragende die Anfrage mit Zeitstempel versieht und die Antwort mit derselben Uhr empfängt, eignet sich Request/Response gut zur Messung der End-to-End-Latenz. Ruf request() ein paar hundert Mal in einer Schleife auf und notiere p50 und p99. Der Round Trip umfasst zwei Broker-Hops, die Verarbeitungszeit des Responders und zwei Netzwerkwege – vergleiche ihn also mit einem Single-Hop-Latenztest, um zu sehen, wie viel Zeit dein Responder hinzufügt. Wiederhole den Test mit QoS 0 und QoS 1: Die zusätzlichen PUBACKs zeigen sich direkt in den Zahlen.

Anfragen und Antworten live beobachten

Request/Response über MQTT 5 braucht nur zwei Properties, aber die Details – vor dem Senden abonnieren, Timeouts, Expiry, eindeutige Correlation-IDs und korrekte ACLs – entscheiden, ob es zuverlässig ist. Behalte die Skripte oben als Regressionssuite und lass sie bei jeder Broker- oder Rechteänderung laufen.

Häufig gestellte Fragen