Cómo probar el Message Expiry Interval de MQTT 5

El Message Expiry Interval de MQTT 5 le dice al broker durante cuánto tiempo un mensaje sigue siendo útil. Si sigue esperando a ser entregado cuando se agota ese tiempo, el broker debe descartarlo en lugar de entregar datos obsoletos. Para probarlo, publica mensajes con distintos valores de expiración mientras un suscriptor persistente está offline, espera, reconecta y comprueba qué mensajes llegan y con qué intervalo restante.

Esta guía recorre esa prueba con mosquitto_pub / mosquitto_sub, un script de Python que se verifica a sí mismo y una prueba aparte para mensajes retenidos.

Cómo funciona realmente la expiración

El publicador fija la propiedad Message Expiry Interval (en segundos) en un paquete PUBLISH. A partir de ahí, la responsabilidad es del broker:

La clave para las pruebas: la expiración solo se nota cuando un mensaje espera. Una prueba con un suscriptor siempre conectado no demuestra nada. Necesitas una sesión offline o un mensaje retenido. Para la teoría y los consejos de diseño, consulta la expiración de mensajes en MQTT.

Requisitos previos

• Un broker con soporte de MQTT 5. Revisa el CONNACK de una conexión MQTT 5, o ejecuta antes las pruebas generales de funciones de MQTT 5.
• Un suscriptor que conserve la sesión mientras está offline: client ID fijo, Clean Start false y un Session Expiry Interval distinto de cero. Si la propia sesión expira, su cola desaparece con ella y tu prueba mide otra cosa. Las sesiones persistentes explican el mecanismo.
• QoS 1 tanto en la suscripción como en la publicación. Muchos brokers no encolan mensajes QoS 0 para clientes offline.

Los ejemplos usan broker.connectmq.com y topics bajo testmqtt/expiry/. Añade un sufijo aleatorio para que tu prueba no choque con la de otros usuarios.

Prueba manual rápida con la CLI de Mosquitto

Paso 1: crea una sesión persistente y suscríbete; después detén el suscriptor con Ctrl+C.

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 desactiva Clean Start, -x 600 mantiene la sesión 10 minutos tras la desconexión y %E imprime el intervalo de expiración de cada mensaje recibido.

Paso 2: mientras el suscriptor está offline, publica tres mensajes:

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'

Paso 3: espera unos 30 segundos y vuelve a ejecutar exactamente el comando del paso 1. Un broker correcto entrega long (con una expiración claramente inferior a 300) y none, y nunca entrega short.

Si llega short, el broker ignora la expiración en los mensajes encolados. Si no llega nada, tu sesión no se conservó: revisa primero la sesión, no la expiración.

Una prueba en Python que se verifica sola

Las pruebas manuales sirven una vez. Para detectar regresiones, usa un script que compruebe el resultado. Este usa paho-mqtt 2.x y propiedades de MQTT 5:

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)

Una ejecución correcta imprime session present: True y un valor de long algo por debajo de 105. Diferencias de un segundo son normales, porque los brokers redondean a segundos enteros.

ABORT con session present: False significa que el broker descartó la sesión o nunca la guardó, así que la expiración no se puede evaluar. Los endpoints públicos compartidos de prueba a menudo no conservan sesiones offline, así que ejecuta esta prueba contra tu propia instancia de broker. Si no, revisa primero la configuración de la sesión: probar la expiración de sesiones cubre este caso.

Probar la expiración de mensajes retenidos

Los mensajes retenidos son el segundo lugar donde importa la expiración, y el que más se olvida. Un estado retenido "dispositivo online" con 30 segundos de expiración desaparece solo si el dispositivo deja de refrescarlo.

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

En la primera salida, %r imprime 1 para un mensaje retenido. En la segunda ejecución no debería imprimirse nada. Si el valor antiguo sigue apareciendo, el broker conserva los retenidos después de su expiración. Algunos brokers borran los retenidos expirados de forma perezosa o con un temporizador, así que deja margen. El comportamiento de los retenidos en general está en probar mensajes retenidos.

También puedes verlo en directo en el cliente MQTT online: suscríbete al topic de estado antes y después de la ventana de expiración y compara.

Ver en el navegador cómo expiran los mensajes

Resultados habituales y qué significan

• El mensaje de vida corta se entrega después de expirar. El broker ignora la propiedad, o la sesión pasa por una ruta de código MQTT 3.1.1. Confirma que publicador y suscriptor se conectan con la versión 5 del protocolo.
• La expiración reenviada es igual al valor original. El broker no resta el tiempo de espera. Eso rompe la lógica posterior que vuelve a reenviar el mensaje, como bridges o gateways.
• Los mensajes recibidos no traen la propiedad, aunque la fijaste. Un bridge o un proxy intermedio puede hablar MQTT 3.1.1 y eliminar las propiedades de MQTT 5.
• No se entrega nada. Revisa session_present, los niveles de QoS (QoS 0 a menudo no se encola) y las ACL. Una suscripción rechazada por el broker nunca encola nada. La consulta online de reason codes te ayuda a descifrar los resultados del SUBACK.
• Los mensajes expiran más tarde de lo esperado. Los brokers comprueban la expiración al entregar o con un barrido periódico. Una diferencia de uno o dos segundos es aceptable; de minutos, no.

Checklist para pruebas de expiración

1. Conecta publicador y suscriptor con MQTT 5 y registra el CONNACK.
2. Usa un client ID fijo, Clean Start false y una expiración de sesión más larga que tu prueba.
3. Publica en QoS 1 con al menos tres expiraciones: menor que la espera, mayor que la espera y ninguna.
4. Mantén el suscriptor offline más tiempo que la expiración corta. Reconecta y comprueba qué llega.
5. Verifica que el intervalo reenviado ha bajado aproximadamente el tiempo de espera.
6. Repite con un mensaje retenido y con un suscriptor nuevo tras la expiración.
7. Limpia: termina la sesión con Session Expiry 0 y borra los topics retenidos con un payload vacío.

La expiración de mensajes es una propiedad pequeña con un gran efecto: decide si un dispositivo que despierta recibe comandos útiles o una ráfaga de comandos obsoletos. Añade el script anterior a tu suite de pruebas y ejecútalo después de cada actualización o migración del broker.

Preguntas frecuentes