Retained Messages testen (und wieder löschen)

Um Retained Messages zu testen, veröffentlichst du mit Retain-Flag, verbindest danach einen völlig neuen Subscriber und prüfst, ob er die Nachricht sofort mit retain=1 erhält. Dann testest du die andere Hälfte des Vertrags: Ein zweiter Retained-Publish überschreibt den ersten, ein Publish ohne Retain nicht, und ein Retained-Publish mit leerem Payload löscht das Topic, sodass der nächste Subscriber nichts bekommt.

Dieser Guide geht jeden Fall mit mosquitto_pub / mosquitto_sub durch, macht daraus eine automatisierte pytest-Suite und behandelt die MQTT-5-Optionen, die das Verhalten von Retained Messages verändern.

Was du prüfen solltest

Die Grundlagen erklärt MQTT Retained Messages. Verwende einen eindeutigen Topic-Baum wie testmqtt/retain-test/..., damit du auf einem geteilten Broker nicht den Zustand anderer liest – und räume danach auf.

Test 1: Ein später Subscriber bekommt den letzten Wert

mosquitto_pub -h broker.connectmq.com -p 1883 \
  -t 'testmqtt/retain-test/door' -m 'closed' -r -q 1

# a brand-new subscriber, started after the publish
mosquitto_sub -h broker.connectmq.com -p 1883 \
  -t 'testmqtt/retain-test/door' -C 1 -W 5 -F '%r %t %p'

Erwartete Ausgabe: 1 testmqtt/retain-test/door closed. -F '%r %t %p' gibt Retain-Flag, Topic und Payload aus; -C 1 beendet nach einer Nachricht und -W 5 gibt nach fünf Sekunden auf, sodass der Befehl in einem Skript nie hängen bleibt. Läuft der Subscriber in den Timeout, wurde die Nachricht nicht gespeichert: Prüfe, ob du wirklich -r übergeben hast, ob deine ACL das Veröffentlichen von Retained Messages erlaubt und ob der Broker Retain überhaupt unterstützt (Retain not supported ist in MQTT 5 der Reason Code 0x9A).

Test 2: Live-Zustellung hat retain=0

Starte zuerst einen Subscriber und lass ihn laufen, dann veröffentliche eine Retained Message:

# Terminal 1: existing subscriber
mosquitto_sub -h broker.connectmq.com -p 1883 -t 'testmqtt/retain-test/#' -F '%r %t %p'

# Terminal 2
mosquitto_pub -h broker.connectmq.com -p 1883 -t 'testmqtt/retain-test/door' -m 'open' -r

Terminal 1 zeigt zuerst den gespeicherten Wert mit 1, dann den neuen mit 0 testmqtt/retain-test/door open. Das ist korrekt: In MQTT 3.1.1 sagt das Flag dem Subscriber „das ist gespeicherter Zustand, den du wegen deiner Subscription bekommst“, nicht „der Publisher hat Retain gesetzt“. Code, der mit msg.retain alten Zustand von Live-Events unterscheidet, funktioniert – Code, der das Flag des Publishers erwartet, nicht. Mit MQTT 5 abonnierst du mit -V mqttv5 --retain-as-published, dann zeigt Terminal 1 auch für die Live-Nachricht 1.

Tests 3–5: überschreiben, ignorieren, löschen

B="-h broker.connectmq.com -p 1883"
T='testmqtt/retain-test/door'

mosquitto_pub $B -t "$T" -m 'open' -r              # 3. overwrite "closed"
mosquitto_pub $B -t "$T" -m 'glitch'               # 4. not retained: must not replace
mosquitto_sub $B -t "$T" -C 1 -W 5 -F '%r %p'      #    → 1 open

mosquitto_pub $B -t "$T" -r -n                     # 5. zero-length retained = clear
mosquitto_sub $B -t "$T" -C 1 -W 3 -F '%r %p'      #    → nothing, times out

-n sendet einen Null-Payload (Länge null). Drei Details bringen Leute regelmäßig aus dem Tritt:

• Das Topic muss exakt übereinstimmen. Eine leere Nachricht an testmqtt/retain-test/# zu veröffentlichen ist ungültig – Wildcards sind in Publish-Topics nicht erlaubt. Lösche jedes Topic einzeln oder nutze die Massenmethode unten.
• Auch die leere Nachricht wird zugestellt. Subscriber, die in dem Moment verbunden sind, erhalten einen Payload der Länge null. Stell sicher, dass deine Handler an b"" nicht abstürzen (ein JSON-Parser tut es).
• Ein Leerzeichen ist nicht leer. -m ' ' speichert eine Retained Message mit einem Byte. Wenn sich ein Topic „nicht löschen lässt“, ist das eine häufige Ursache – siehe Retained Message lässt sich nicht löschen.

Test 6: Wildcards und Massenbereinigung

# list: prints only retained messages, stops at the first live one or after 3 s
mosquitto_sub -h broker.connectmq.com -p 1883 \
  -t 'testmqtt/retain-test/#' --retained-only -W 3 -F '%t = %p'

# remove: publishes an empty retained message for every retained topic it receives
mosquitto_sub -h broker.connectmq.com -p 1883 \
  -t 'testmqtt/retain-test/#' --retained-only --remove-retained -W 3

Eine Wildcard-Subscription erhält eine Retained Message pro passendem Topic. Ein breiter Filter kann nach einem Deployment also Tausende Nachrichten auf einmal liefern. Prüfe mit dem Topic Matcher, welche Topics ein Filter abdeckt, bevor du --remove-retained ausführst – auf einem Produktions-Broker löscht es alles, was passt.

Automatisierung mit pytest

Derselbe Vertrag als automatisierte Suite (paho-mqtt 2.x). Jeder Test bekommt sein eigenes zufälliges Topic, und die Fixture räumt es danach auf:

import threading, time, uuid
import paho.mqtt.client as mqtt
import pytest

HOST, PORT = "broker.connectmq.com", 1883

def connect():
    ready = threading.Event()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    # c.username_pw_set("username", "password")
    c.on_connect = lambda cl, ud, flags, rc, props: ready.set()
    c.connect(HOST, PORT)
    c.loop_start()
    assert ready.wait(5), "no CONNACK"
    return c

def publish(topic, payload, retain=True):
    c = connect()
    c.publish(topic, payload, qos=1, retain=retain).wait_for_publish(5)
    c.disconnect()
    c.loop_stop()

def fetch(topic, wait=1.5):
    """Subscribe with a brand-new client; return [(payload, retain_flag), ...]."""
    got = []
    c = connect()
    c.on_message = lambda cl, ud, msg: got.append((msg.payload, msg.retain))
    c.subscribe(topic, qos=1)
    time.sleep(wait)
    c.disconnect()
    c.loop_stop()
    return got

@pytest.fixture
def topic():
    t = f"testmqtt/retain-test/{uuid.uuid4().hex[:8]}"
    yield t
    publish(t, b"")                     # always clean up

def test_late_subscriber_gets_retained(topic):
    publish(topic, b"closed")
    assert fetch(topic) == [(b"closed", True)]

def test_overwrite_keeps_only_last(topic):
    publish(topic, b"one")
    publish(topic, b"two")
    assert fetch(topic) == [(b"two", True)]

def test_non_retained_publish_does_not_replace(topic):
    publish(topic, b"closed")
    publish(topic, b"glitch", retain=False)
    assert fetch(topic) == [(b"closed", True)]

def test_empty_payload_clears(topic):
    publish(topic, b"closed")
    publish(topic, b"")
    assert fetch(topic) == []

Das feste wait hält das Beispiel kurz; in CI ersetzt du es durch ein Event, das bei der ersten Nachricht auslöst, plus eine kurze Karenzzeit. Wiederverwendbare Broker-Fixtures findest du unter pytest-Fixtures für MQTT.

MQTT-5-Optionen, die das Ergebnis ändern

MQTT 5 bringt Subscription-Optionen mit, die ein vollständiger Test abdecken sollte (Details unter Subscription Options):

• Retain Handling – 0 sendet Retained Messages bei jedem Subscribe (Standard), 1 nur bei einer neuen Subscription, 2 nie.
• Retain As Published – behält das Retain-Flag des Publishers bei Live-Zustellungen (Test 2 zeigt dann 1).
• Message Expiry Interval – eine Retained Message mit Ablaufzeit wird entfernt, sobald sie abläuft; teste das mit den Message-Expiry-Tests.

import paho.mqtt.client as mqtt
from paho.mqtt.subscribeoptions import SubscribeOptions

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
c.on_message = lambda cl, ud, msg: print("got", msg.topic, msg.payload, msg.retain)
c.connect("broker.connectmq.com", 1883)
c.subscribe("testmqtt/retain-test/door",
            options=SubscribeOptions(qos=1, retainHandling=SubscribeOptions.RETAIN_DO_NOT_SEND))
c.loop_forever()   # prints only messages published from now on

Test 7: Übersteht der Retained-Zustand einen Neustart?

Retained Messages liegen im Speicher des Brokers, sofern er sie nicht auf die Platte schreibt. Bei einem selbst gehosteten Broker veröffentlichst du einen Retained-Wert, startest den Broker neu und führst Test 1 erneut aus. Ist der Wert weg, aktiviere die Persistenz (bei Mosquitto: persistence true und eine persistence_location) oder akzeptiere, dass Geräte ihren Zustand nach einem Broker-Neustart erneut veröffentlichen müssen. Managed Broker kümmern sich darum, aber es lohnt sich trotzdem, den Test einmal laufen zu lassen.

Retained Messages im Browser untersuchen

Retained Messages sind ein kleiner Vertrag: gespeichert beim Publish mit Retain, einmal pro neuer Subscription zugestellt, vom nächsten Retained-Publish ersetzt und von einem leeren gelöscht. Teste jede Klausel einmal von Hand mit dem Online-MQTT-Client oder mosquitto_sub -F '%r' und behalte die pytest-Suite, damit ein Broker-Wechsel nie stillschweigend den „letzten bekannten Zustand“ deiner Geräte kaputt macht.

Häufig gestellte Fragen