MQTT 5 Message Expiry Interval testen

Das Message Expiry Interval von MQTT 5 sagt dem Broker, wie lange eine Nachricht nützlich bleibt. Wartet sie nach Ablauf dieser Zeit noch auf Zustellung, muss der Broker sie verwerfen, statt veraltete Daten auszuliefern. Zum Testen veröffentlichst du Nachrichten mit unterschiedlichen Expiry-Werten, während ein persistenter Subscriber offline ist, wartest, verbindest neu und prüfst, welche Nachrichten mit welchem Restintervall ankommen.

Dieser Guide zeigt den Test mit mosquitto_pub / mosquitto_sub, einem selbstprüfenden Python-Skript und einem eigenen Test für Retained Messages.

Wie Message Expiry wirklich funktioniert

Der Publisher setzt die Property Message Expiry Interval (in Sekunden) auf ein PUBLISH-Paket. Ab dann ist der Broker zuständig:

Der Kern für Tests: Expiry wird erst sichtbar, wenn eine Nachricht wartet. Ein Test mit einem ständig verbundenen Subscriber beweist nichts. Du brauchst eine Offline-Session oder eine Retained Message. Theorie und Design-Tipps findest du unter MQTT Message Expiry.

Voraussetzungen

• Ein Broker mit MQTT-5-Unterstützung. Prüfe das CONNACK einer MQTT-5-Verbindung oder führe vorher die allgemeinen MQTT-5-Funktionstests aus.
• Ein Subscriber, der seine Session offline behält: feste Client-ID, Clean Start false und ein Session Expiry Interval größer null. Läuft die Session selbst ab, verschwindet ihre Queue mit, und dein Test misst das Falsche. Persistente Sessions erklärt die Mechanik.
• QoS 1 bei Subscription und Publish. Viele Broker reihen QoS-0-Nachrichten für Offline-Clients gar nicht ein.

Die Beispiele nutzen broker.connectmq.com und Topics unter testmqtt/expiry/. Hänge ein zufälliges Suffix an, damit dein Test nicht mit anderen Nutzern kollidiert.

Schneller manueller Test mit der Mosquitto-CLI

Schritt 1: Persistente Session anlegen und abonnieren, dann den Subscriber mit Strg+C beenden.

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -i expiry-sub-7f3a -c -x 600 -q 1 \
  -t 'testmqtt/expiry/7f3a/#' -F '%t expiry=%E payload=%p'

-c deaktiviert Clean Start, -x 600 hält die Session 10 Minuten nach dem Disconnect, und %E gibt das Expiry Interval jeder empfangenen Nachricht aus.

Schritt 2: Während der Subscriber offline ist, drei Nachrichten veröffentlichen:

mosquitto_pub -h broker.connectmq.com -V mqttv5 -q 1 \
  -t testmqtt/expiry/7f3a/short -m 'expires in 10s' \
  -D publish message-expiry-interval 10

mosquitto_pub -h broker.connectmq.com -V mqttv5 -q 1 \
  -t testmqtt/expiry/7f3a/long -m 'expires in 300s' \
  -D publish message-expiry-interval 300

mosquitto_pub -h broker.connectmq.com -V mqttv5 -q 1 \
  -t testmqtt/expiry/7f3a/none -m 'never expires'

Schritt 3: Etwa 30 Sekunden warten und genau den Befehl aus Schritt 1 erneut ausführen. Ein korrekter Broker liefert long (mit einem Expiry deutlich unter 300) und none, aber niemals short.

Kommt short an, ignoriert der Broker Expiry für eingereihte Nachrichten. Kommt gar nichts an, wurde deine Session nicht gehalten: Prüfe zuerst die Session, nicht Expiry.

Ein selbstprüfender Python-Test

Manuelle Tests reichen einmal. Für Regressionen brauchst du ein Skript, das das Ergebnis prüft. Dieses nutzt paho-mqtt 2.x und MQTT-5-Properties:

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

HOST, PORT = "broker.connectmq.com", 1883
RUN = uuid.uuid4().hex[:8]
TOPIC = f"testmqtt/expiry/{RUN}"
SUB_ID = f"expiry-sub-{RUN}"
WAIT = 15                                  # seconds the subscriber stays offline
received = {}
session = {}

def client(cid):
    return mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=cid, protocol=mqtt.MQTTv5)

def connect_props(session_expiry):
    p = Properties(PacketTypes.CONNECT)
    p.SessionExpiryInterval = session_expiry
    return p

def on_message(c, userdata, msg):
    left = getattr(msg.properties, "MessageExpiryInterval", None)
    received[msg.topic.rsplit("/", 1)[1]] = left

# 1. Create a persistent session with a subscription, then go offline
sub = client(SUB_ID)
sub.connect(HOST, PORT, clean_start=True, properties=connect_props(600))
sub.loop_start()
sub.subscribe(f"{TOPIC}/#", qos=1)
time.sleep(1)
sub.disconnect()
sub.loop_stop()

# 2. Publish with different expiry intervals while it is offline
pub = client(f"expiry-pub-{RUN}")
pub.connect(HOST, PORT)
pub.loop_start()
for name, expiry in [("short", 5), ("long", 120), ("none", None)]:
    props = Properties(PacketTypes.PUBLISH)
    if expiry:
        props.MessageExpiryInterval = expiry
    pub.publish(f"{TOPIC}/{name}", name, qos=1, properties=props).wait_for_publish()
pub.disconnect()
pub.loop_stop()

time.sleep(WAIT)

# 3. Resume the session and collect what is left
sub = client(SUB_ID)
sub.on_message = on_message
sub.on_connect = lambda c, u, flags, rc, p: session.update(present=flags.session_present)
sub.connect(HOST, PORT, clean_start=False, properties=connect_props(600))
sub.loop_start()
time.sleep(3)
end = Properties(PacketTypes.DISCONNECT)
end.SessionExpiryInterval = 0              # clean up the session
sub.disconnect(properties=end)
sub.loop_stop()

print("session present:", session.get("present"), "received:", received)
if not session.get("present"):
    raise SystemExit("ABORT: the broker did not keep the session, expiry cannot be judged")
checks = {
    "short expired": "short" not in received,
    "long delivered": "long" in received,
    "long shows remaining time": received.get("long") is not None and received["long"] <= 120 - WAIT,
    "no-expiry delivered": "none" in received,
}
for name, ok in checks.items():
    print(("PASS " if ok else "FAIL ") + name)

Ein erfolgreicher Lauf gibt session present: True und einen long-Wert knapp unter 105 aus. Abweichungen von einer Sekunde sind normal, weil Broker auf ganze Sekunden runden.

ABORT mit session present: False bedeutet, dass der Broker die Session verworfen oder nie gespeichert hat, Expiry lässt sich dann nicht beurteilen. Geteilte öffentliche Test-Endpoints behalten Offline-Sessions oft gar nicht, also führe diesen Test gegen deine eigene Broker-Instanz aus. Andernfalls prüfe zuerst die Session-Einstellungen: Session Expiry testen behandelt genau diesen Fall.

Ablauf von Retained Messages testen

Retained Messages sind der zweite Ort, an dem Expiry zählt, und der, den man vergisst. Ein retained Status "Gerät online" mit 30 Sekunden Expiry verschwindet automatisch, wenn das Gerät ihn nicht mehr erneuert.

mosquitto_pub -h broker.connectmq.com -V mqttv5 -q 1 -r \
  -t testmqtt/expiry/7f3a/status -m online \
  -D publish message-expiry-interval 20

# Immediately: prints the message (expiry close to 20)
mosquitto_sub -h broker.connectmq.com -V mqttv5 \
  -t testmqtt/expiry/7f3a/status -F '%r %E %p' -W 3

sleep 25

# After expiry: prints nothing and exits after the 3 s timeout
mosquitto_sub -h broker.connectmq.com -V mqttv5 \
  -t testmqtt/expiry/7f3a/status -F '%r %E %p' -W 3

In der ersten Ausgabe druckt %r eine 1 für eine Retained Message. Im zweiten Lauf sollte nichts erscheinen. Taucht der alte Wert weiterhin auf, behält der Broker Retained Messages über ihren Ablauf hinaus. Manche Broker löschen abgelaufene Retained Messages verzögert oder per Timer, teste also mit etwas Puffer. Retained-Verhalten allgemein beschreibt Retained Messages testen.

Live beobachten kannst du das auch im Online-MQTT-Client: Abonniere das Status-Topic vor und nach dem Ablauffenster und vergleiche.

Ablaufende Nachrichten im Browser beobachten

Typische Ergebnisse und ihre Bedeutung

• Die kurzlebige Nachricht wird nach ihrem Ablauf zugestellt. Der Broker ignoriert die Property, oder die Session läuft über einen MQTT-3.1.1-Codepfad. Prüfe, dass Publisher und Subscriber mit Protokollversion 5 verbinden.
• Das weitergeleitete Expiry entspricht dem Originalwert. Der Broker zieht die Wartezeit nicht ab. Das bricht nachgelagerte Logik, die die Nachricht erneut weiterleitet, etwa Bridges oder Gateways.
• Empfangene Nachrichten haben keine Property, obwohl Expiry gesetzt war. Eine Bridge oder ein Proxy dazwischen spricht vielleicht MQTT 3.1.1 und entfernt MQTT-5-Properties.
• Es kommt überhaupt nichts an. Prüfe session_present, die QoS-Stufen (QoS 0 wird oft nicht eingereiht) und ACLs. Eine vom Broker abgelehnte Subscription reiht nie etwas ein. Die Reason-Code-Suche hilft beim Entschlüsseln von SUBACK-Ergebnissen.
• Nachrichten laufen später ab als erwartet. Broker prüfen Expiry bei der Zustellung oder in einem periodischen Durchlauf. Ein bis zwei Sekunden Differenz sind in Ordnung, Minuten nicht.

Checkliste für Expiry-Tests

1. Publisher und Subscriber mit MQTT 5 verbinden und das CONNACK loggen.
2. Feste Client-ID, Clean Start false und eine Session Expiry länger als der Test verwenden.
3. Mit QoS 1 und mindestens drei Expiry-Werten veröffentlichen: kürzer als die Wartezeit, länger als die Wartezeit und ohne.
4. Den Subscriber länger offline lassen als das kurze Expiry. Neu verbinden und prüfen, was ankommt.
5. Kontrollieren, dass das weitergeleitete Intervall ungefähr um die Wartezeit gesunken ist.
6. Mit einer Retained Message und einem neuen Subscriber nach Ablauf wiederholen.
7. Aufräumen: Session mit Session Expiry 0 beenden und Retained Topics mit leerem Payload löschen.

Message Expiry ist eine kleine Property mit großer Wirkung: Sie entscheidet, ob ein aufwachendes Gerät nützliche Befehle bekommt oder eine Flut veralteter. Nimm das Skript oben in deine Test-Suite auf und führe es nach jedem Broker-Upgrade oder jeder Migration aus.

Häufig gestellte Fragen