Record and Replay MQTT Traffic for Testing

Recording and replaying MQTT traffic means capturing real messages from a broker to a file — topic, payload, QoS, retain flag and the time each one arrived — and later publishing that file back with the same timing. It gives you regression tests built from what your devices actually send instead of what someone assumed they send.

This guide builds a small recorder and a timing-accurate replayer in Python (paho-mqtt 2.x), then shows how to turn a recording into an automated regression test.

Why replay beats synthetic data

Generated test data is clean. Real fleets are not: firmware versions send slightly different JSON, sensors burst ten readings in 50 ms after a reconnect, a gateway republishes the same retained status every minute. Bugs live in those details.

Synthetic data still has its place — use a test data generator for edge cases nobody has recorded yet — but a recording is the closest thing to production you can run on a laptop.

Choose what to record

A recorder is an ordinary MQTT client with a subscription. Pick a topic filter that covers the flow you want to test, not # on the whole broker: smaller captures are easier to review, and they leak less data. Check a filter before you start with the topic matcher.

Things to decide up front:

• Topic filter — for example testmqtt/replay/site1/# for one site's telemetry.
• QoS of the subscription — subscribe with QoS 2. Delivery QoS is the lower of the publish QoS and the subscription QoS, so a QoS 2 subscription preserves the original level.
• Payload encoding — store payloads as base64 so binary formats (Protobuf, CBOR) survive the round trip.
• Privacy — real payloads may contain locations, names or tokens. Scrub them before committing a capture to a repository.

A recorder in 30 lines

The recorder writes one JSON object per line (JSONL). The t field is seconds since the recorder started, measured with a monotonic clock so system clock adjustments cannot distort the timing.

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("connected:", reason_code)
    client.subscribe(FILTER, qos=2)        # resubscribes after every reconnect

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()

Run it with python record.py 'testmqtt/replay/#' capture.jsonl, generate some traffic, then press Ctrl+C. Line buffering (buffering=1) means the file is usable even if the process is killed.

Retained messages in a capture

When the recorder subscribes, the broker immediately sends every matching retained message with retain: true. These are a snapshot of state at recording time, not live traffic, and they all arrive at t ≈ 0. Messages forwarded live arrive with retain: false, even when the publisher set the flag (MQTT 5 clients can change this with the Retain As Published subscription option). The replayer below skips the snapshot by default; publish it separately if your test needs initial state.

A replayer that keeps the timing

The naive approach — sleep(gap) between messages — drifts, because every publish call takes time too. Instead, compute the due time of each message from the start of the replay and sleep only until then. A speed factor lets you compress an hour of traffic into minutes.

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 = twice as fast
SRC, DST = "testmqtt/replay/", "testmqtt/replay-run/"          # topic rewrite

records = [json.loads(line) for line in open(FILE) if line.strip()]
records = [r for r in records if not r["retain"]]              # skip retained snapshot
if not records:
    sys.exit("nothing to replay")
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"replayed {len(records)} messages, worst lag {late * 1000:.1f} ms")
client.loop_stop()
client.disconnect()

The topic rewrite is the safety net: the recording came from testmqtt/replay/..., the replay goes to testmqtt/replay-run/..., so nothing that listens to the original topics reacts to it. Point the system under test at the rewritten prefix.

The worst lag line tells you whether the replay was faithful. If it grows at high speed factors, the replayer itself is the bottleneck — the client's inflight window is full of unacknowledged QoS 1/2 messages. See client performance tuning for raising it.

Turn a replay into a regression test

A replay becomes a test when you compare what the system under test produced against a known-good result. The pattern:

1. Record the inputs (device telemetry) once and commit the capture, scrubbed, as a fixture.
2. Run the recorder on the system's output topics — alerts, aggregates, commands.
3. Replay the inputs.
4. Compare the output capture with a golden file recorded from a trusted build.

Timing is never identical between runs, so compare content, not timestamps. Group by topic and compare the sequence of payloads per topic; MQTT guarantees order only per topic and per publisher, so a global order comparison produces flaky 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}: expected {len(golden[topic])} msgs, got {len(actual[topic])}")
print("PASS" if ok else "FAIL")
sys.exit(0 if ok else 1)

If payloads contain timestamps or generated IDs, normalize them before comparing (parse the JSON and drop those fields). Wire the three scripts into a pytest fixture or a CI job — the integration testing guide and MQTT testing in GitHub Actions show how to start a broker for each run.

Pitfalls to watch

• Replaying commands. A capture of .../cmd/... topics will switch real relays if it reaches a real device. Filter command topics out of input captures unless the test is about them.
• Duplicate messages. QoS 1 can deliver a message twice after a reconnect, and the recording keeps the duplicate. That is realistic — your consumers should handle it — but know it is there before you count messages.
• Missing MQTT 5 metadata. The scripts above store topic, payload, QoS and retain. If your system relies on user properties, content type or response topics, add msg.properties fields to the record and pass a Properties object when publishing.
• Clock-dependent consumers. Logic that uses wall-clock time (for example "no reading for 60 s") behaves differently at 10× speed. Replay those tests at 1×.
• Broker limits. Fast replays of large captures can hit message rate or queue limits on shared brokers. Use your own instance for anything beyond a quick check.

Check the replay by eye

Before automating anything, watch one replay arrive. Open the online MQTT client, connect to broker.connectmq.com over secure WebSockets and subscribe to the rewritten prefix, then start replay.py. You should see messages arrive in bursts that match the original pattern.

Watch the replay in your browser

A recorder, a replayer and a comparison script are fewer than a hundred lines of Python, and they turn real traffic into repeatable tests. Record once from the field, scrub the capture, keep it next to your code, and every future change gets tested against what your devices really send.

Frequently asked questions