MQTT-Traffic für Tests aufzeichnen und abspielen

MQTT-Traffic aufzeichnen und abspielen heißt: echte Nachrichten eines Brokers in eine Datei mitschneiden — Topic, Payload, QoS, Retain-Flag und den Zeitpunkt, zu dem jede ankam — und diese Datei später mit demselben Timing wieder publizieren. So entstehen Regressionstests aus dem, was deine Geräte tatsächlich senden, statt aus dem, was jemand vermutet hat.

Diese Anleitung baut einen kleinen Recorder und einen zeitgenauen Replayer in Python (paho-mqtt 2.x) und zeigt dann, wie aus einer Aufzeichnung ein automatisierter Regressionstest wird.

Warum Replay besser ist als synthetische Daten

Generierte Testdaten sind sauber. Echte Flotten nicht: Firmware-Versionen senden leicht unterschiedliches JSON, Sensoren schicken nach einem Reconnect zehn Messwerte in 50 ms, ein Gateway publiziert jede Minute denselben Retained-Status neu. Genau in diesen Details stecken die Bugs.

Synthetische Daten haben trotzdem ihren Platz — nutze einen Testdaten-Generator für Grenzfälle, die noch niemand aufgezeichnet hat —, aber eine Aufzeichnung ist das Produktionsnächste, was du auf einem Laptop laufen lassen kannst.

Was aufzeichnen?

Ein Recorder ist ein gewöhnlicher MQTT-Client mit einer Subscription. Wähle einen Topic-Filter, der den zu testenden Ablauf abdeckt, nicht # auf dem ganzen Broker: Kleinere Mitschnitte lassen sich leichter prüfen und geben weniger Daten preis. Prüfe den Filter vorher mit dem Topic Matcher.

Vorab zu entscheiden:

• Topic-Filter — zum Beispiel testmqtt/replay/site1/# für die Telemetrie eines Standorts.
• QoS der Subscription — abonniere mit QoS 2. Die Zustell-QoS ist das Minimum aus Publish-QoS und Subscription-QoS, eine QoS-2-Subscription erhält also das ursprüngliche Level.
• Payload-Kodierung — speichere Payloads als Base64, damit Binärformate (Protobuf, CBOR) den Hin- und Rückweg überstehen.
• Datenschutz — echte Payloads können Standorte, Namen oder Tokens enthalten. Bereinige sie, bevor du einen Mitschnitt in ein Repository committest.

Ein Recorder in 30 Zeilen

Der Recorder schreibt pro Zeile ein JSON-Objekt (JSONL). Das Feld t enthält die Sekunden seit dem Start des Recorders, gemessen mit einer monotonen Uhr, damit Anpassungen der Systemzeit das Timing nicht verfälschen.

import base64, json, os, sys, time
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
FILTER = sys.argv[1] if len(sys.argv) > 1 else "testmqtt/replay/#"
out = open(sys.argv[2] if len(sys.argv) > 2 else "capture.jsonl", "w", buffering=1)
start = time.monotonic()

def on_connect(client, userdata, flags, reason_code, properties):
    print("verbunden:", reason_code)
    client.subscribe(FILTER, qos=2)        # abonniert nach jedem Reconnect erneut

def on_message(client, userdata, msg):
    out.write(json.dumps({
        "t": round(time.monotonic() - start, 6),
        "topic": msg.topic,
        "qos": msg.qos,
        "retain": msg.retain,
        "payload": base64.b64encode(msg.payload).decode(),
    }) + "\n")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"testmqtt-recorder-{os.getpid()}")
if os.environ.get("MQTT_USER"):
    client.username_pw_set(os.environ["MQTT_USER"], os.environ.get("MQTT_PASS"))
client.on_connect = on_connect
client.on_message = on_message
client.connect(HOST, PORT, keepalive=30)
try:
    client.loop_forever()
except KeyboardInterrupt:
    client.disconnect()
    out.close()

Starte ihn mit python record.py 'testmqtt/replay/#' capture.jsonl, erzeuge etwas Traffic und drücke Ctrl+C. Durch Zeilenpufferung (buffering=1) bleibt die Datei auch dann nutzbar, wenn der Prozess abgeschossen wird.

Retained Messages im Mitschnitt

Sobald der Recorder abonniert, schickt der Broker sofort alle passenden Retained Messages mit retain: true. Sie sind eine Momentaufnahme des Zustands zum Aufnahmezeitpunkt, kein Live-Traffic, und kommen alle bei t ≈ 0 an. Live weitergeleitete Nachrichten kommen mit retain: false an, selbst wenn der Publisher das Flag gesetzt hat (MQTT-5-Clients können das mit der Subscription-Option Retain As Published ändern). Der Replayer unten überspringt die Momentaufnahme standardmäßig; publiziere sie separat, wenn dein Test einen Anfangszustand braucht.

Ein Replayer, der das Timing einhält

Der naive Ansatz — sleep(gap) zwischen Nachrichten — driftet, weil auch jeder Publish-Aufruf Zeit kostet. Berechne stattdessen den Sollzeitpunkt jeder Nachricht ab Start des Replays und schlafe nur bis dahin. Ein Geschwindigkeitsfaktor komprimiert eine Stunde Traffic auf Minuten.

import base64, json, os, sys, time
import paho.mqtt.client as mqtt

FILE = sys.argv[1]
SPEED = float(sys.argv[2]) if len(sys.argv) > 2 else 1.0     # 2.0 = doppelt so schnell
SRC, DST = "testmqtt/replay/", "testmqtt/replay-run/"          # Topic-Umschreibung

records = [json.loads(line) for line in open(FILE) if line.strip()]
records = [r for r in records if not r["retain"]]              # Retained-Snapshot überspringen
if not records:
    sys.exit("nichts abzuspielen")
offset = records[0]["t"]

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"testmqtt-replayer-{os.getpid()}")
if os.environ.get("MQTT_USER"):
    client.username_pw_set(os.environ["MQTT_USER"], os.environ.get("MQTT_PASS"))
client.connect("broker.connectmq.com", 1883)
client.loop_start()

t0 = time.monotonic()
late = 0.0
for r in records:
    due = t0 + (r["t"] - offset) / SPEED
    delay = due - time.monotonic()
    if delay > 0:
        time.sleep(delay)
    else:
        late = max(late, -delay)
    topic = r["topic"].replace(SRC, DST, 1)
    info = client.publish(topic, base64.b64decode(r["payload"]), qos=r["qos"])

info.wait_for_publish(timeout=30)
print(f"{len(records)} Nachrichten abgespielt, größte Verspätung {late * 1000:.1f} ms")
client.loop_stop()
client.disconnect()

Die Topic-Umschreibung ist das Sicherheitsnetz: Die Aufzeichnung stammt von testmqtt/replay/..., das Replay geht nach testmqtt/replay-run/..., also reagiert nichts, was auf den Original-Topics lauscht. Richte das System unter Test auf das umgeschriebene Präfix aus.

Die Zeile mit der größten Verspätung zeigt, ob das Replay originalgetreu war. Wächst sie bei hohen Geschwindigkeitsfaktoren, ist der Replayer selbst der Engpass — das Inflight-Fenster des Clients ist voller unbestätigter QoS-1/2-Nachrichten. Wie du es vergrößerst, steht im Client-Performance-Tuning.

Aus einem Replay einen Regressionstest machen

Ein Replay wird zum Test, wenn du vergleichst, was das System unter Test erzeugt hat, mit einem bekannt guten Ergebnis. Das Muster:

1. Zeichne die Eingaben (Geräte-Telemetrie) einmal auf und committe den bereinigten Mitschnitt als Fixture.
2. Lass den Recorder auf den Ausgabe-Topics des Systems laufen — Alarme, Aggregate, Befehle.
3. Spiele die Eingaben ab.
4. Vergleiche den Ausgabe-Mitschnitt mit einer Golden-Datei aus einem vertrauenswürdigen Build.

Das Timing ist zwischen Läufen nie identisch, vergleiche also Inhalte, keine Zeitstempel. Gruppiere nach Topic und vergleiche die Payload-Folge pro Topic; MQTT garantiert Reihenfolge nur pro Topic und Publisher, ein globaler Reihenfolgevergleich erzeugt wackelige Tests.

import base64, json, sys
from collections import defaultdict

def load(path):
    per_topic = defaultdict(list)
    for line in open(path):
        if line.strip():
            r = json.loads(line)
            per_topic[r["topic"]].append(base64.b64decode(r["payload"]))
    return per_topic

golden, actual = load(sys.argv[1]), load(sys.argv[2])
ok = True
for topic in sorted(set(golden) | set(actual)):
    if golden[topic] != actual[topic]:
        ok = False
        print(f"DIFF {topic}: erwartet {len(golden[topic])} msgs, erhalten {len(actual[topic])}")
print("PASS" if ok else "FAIL")
sys.exit(0 if ok else 1)

Enthalten Payloads Zeitstempel oder generierte IDs, normalisiere sie vor dem Vergleich (JSON parsen, diese Felder entfernen). Binde die drei Skripte in ein pytest-Fixture oder einen CI-Job ein — der Leitfaden zu Integrationstests und MQTT-Tests in GitHub Actions zeigen, wie du pro Lauf einen Broker startest.

Fallstricke

• Befehle abspielen. Ein Mitschnitt von .../cmd/...-Topics schaltet echte Relais, wenn er ein echtes Gerät erreicht. Filtere Befehls-Topics aus Eingabe-Mitschnitten heraus, sofern der Test nicht genau davon handelt.
• Doppelte Nachrichten. QoS 1 kann eine Nachricht nach einem Reconnect zweimal zustellen, und die Aufzeichnung enthält das Duplikat. Das ist realistisch — deine Consumer sollten damit umgehen —, aber denk daran, bevor du Nachrichten zählst.
• Fehlende MQTT-5-Metadaten. Die Skripte oben speichern Topic, Payload, QoS und Retain. Hängt dein System an User Properties, Content Type oder Response Topics, ergänze Felder aus msg.properties und übergib beim Publish ein Properties-Objekt.
• Uhrzeitabhängige Consumer. Logik mit Wanduhrzeit (etwa „kein Messwert seit 60 s“) verhält sich bei 10-facher Geschwindigkeit anders. Spiele solche Tests mit 1× ab.
• Broker-Limits. Schnelle Replays großer Mitschnitte können auf geteilten Brokern Nachrichtenraten- oder Queue-Limits treffen. Nutze für alles über einen kurzen Check hinaus eine eigene Instanz.

Das Replay mit eigenen Augen prüfen

Bevor du etwas automatisierst, schau einem Replay beim Ankommen zu. Öffne den Online-MQTT-Client, verbinde dich über Secure WebSockets mit broker.connectmq.com, abonniere das umgeschriebene Präfix und starte dann replay.py. Die Nachrichten sollten in Schüben ankommen, die dem Originalmuster entsprechen.

Replay im Browser verfolgen

Recorder, Replayer und Vergleichsskript sind zusammen weniger als hundert Zeilen Python und machen aus echtem Traffic wiederholbare Tests. Einmal im Feld aufzeichnen, Mitschnitt bereinigen, neben dem Code ablegen — und jede künftige Änderung wird gegen das getestet, was deine Geräte wirklich senden.

Häufig gestellte Fragen