Probar mensajes retenidos MQTT y cómo borrarlos

Para probar los mensajes retenidos, publica con el flag retain, conecta después un suscriptor completamente nuevo y comprueba que recibe el mensaje al instante con retain=1. Luego prueba la otra mitad del contrato: un segundo publish retenido sobrescribe al primero, un publish no retenido no lo hace, y un publish retenido con payload vacío borra el topic, de modo que el siguiente suscriptor no recibe nada.

Esta guía recorre cada caso con mosquitto_pub / mosquitto_sub, los convierte en una suite automatizada con pytest y cubre las opciones de MQTT 5 que cambian el comportamiento de los mensajes retenidos.

Qué verificar

#	Prueba	Resultado esperado
1	Publish retenido y luego un suscriptor nuevo	Mensaje entregado al momento, flag retain 1
2	Un suscriptor ya conectado recibe un publish retenido	Entregado, flag retain 0 (por defecto en MQTT 3.1.1)
3	Segundo publish retenido en el mismo topic	Solo se guarda el último valor
4	Publish no retenido en el mismo topic	El valor guardado no cambia
5	Publish retenido con payload vacío	Topic borrado; los nuevos suscriptores no reciben nada
6	Suscripción con wildcard sobre topics retenidos	Un mensaje retenido por cada topic que coincide
7	Reinicio del broker	Los retenidos sobreviven solo si el broker los persiste

Los fundamentos se explican en mensajes retenidos de MQTT. Usa un árbol de topics único, como testmqtt/retain-test/..., para no leer el estado de otra persona en un broker compartido, y límpialo al terminar.

Prueba 1: un suscriptor tardío recibe el último valor

Publish once, subscribe afterwards

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'

Salida esperada: 1 testmqtt/retain-test/door closed. -F '%r %t %p' imprime el flag retain, el topic y el payload; -C 1 termina tras un mensaje y -W 5 se rinde a los cinco segundos, así que el comando nunca se queda colgado en un script. Si el suscriptor agota el tiempo, el mensaje no se guardó: comprueba que realmente pasaste -r, que tu ACL permite publicar mensajes retenidos y si el broker admite retain (Retain not supported es el reason code 0x9A en MQTT 5).

Prueba 2: la entrega en vivo lleva retain=0

Arranca primero un suscriptor y déjalo en marcha; después publica un mensaje retenido:

Two terminals

# 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

La terminal 1 imprime primero el valor guardado con 1 y después el nuevo con 0 testmqtt/retain-test/door open. Es correcto: en MQTT 3.1.1 el flag le dice al suscriptor "esto es estado almacenado que recibes porque te suscribiste", no "el publicador activó retain". El código que usa msg.retain para distinguir estado antiguo de eventos en vivo funciona; el código que espera ver el flag del publicador, no. Con MQTT 5, suscríbete con -V mqttv5 --retain-as-published y la terminal 1 mostrará 1 también para el mensaje en vivo.

Pruebas 3–5: sobrescribir, ignorar, borrar

Overwrite and clear

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 envía un payload nulo (de longitud cero). Tres detalles suelen confundir:

El topic debe coincidir exactamente. Publicar un mensaje vacío en testmqtt/retain-test/# no es válido: los wildcards no se permiten en topics de publicación. Borra cada topic o usa el método masivo de abajo.
El mensaje vacío también se entrega. Los suscriptores conectados en ese momento reciben un payload de longitud cero. Asegúrate de que tus manejadores no fallen con b"" (un parser JSON fallará).
Un espacio no es vacío. -m ' ' guarda un mensaje retenido de un byte. Si un topic "no se borra", esta es una causa habitual: consulta el mensaje retenido no se borra.

Prueba 6: wildcards y limpieza masiva

List and remove every retained message under a prefix

# 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

Una suscripción con wildcard recibe un mensaje retenido por cada topic que coincide, así que suscribirse a un filtro amplio tras un despliegue puede entregar miles de mensajes de golpe. Comprueba qué topics cubre un filtro con el topic matcher antes de ejecutar --remove-retained: en un broker de producción borra todo lo que coincida.

Automatizarlo con pytest

El mismo contrato como suite automatizada (paho-mqtt 2.x). Cada prueba recibe su propio topic aleatorio y el fixture lo borra al final:

test_retained.py

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) == []

El wait fijo mantiene corto el ejemplo; en CI, sustitúyelo por un evento que se dispare con el primer mensaje más un breve margen. Para fixtures de broker reutilizables consulta fixtures de pytest para MQTT.

Opciones de MQTT 5 que cambian el resultado

MQTT 5 añade opciones de suscripción que una prueba completa debería cubrir (detalles en opciones de suscripción):

Retain Handling: 0 envía los retenidos en cada subscribe (por defecto), 1 solo cuando la suscripción es nueva, 2 nunca.
Retain As Published: conserva el flag retain del publicador en las entregas en vivo (la prueba 2 muestra entonces 1).
Message Expiry Interval: un mensaje retenido con expiración se elimina cuando expira; pruébalo con las pruebas de expiración de mensajes.

Retain Handling = 2: no stored messages on subscribe

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

Prueba 7: ¿sobrevive el estado retenido a un reinicio?

Los mensajes retenidos viven en la memoria del broker salvo que este los persista en disco. En un broker autoalojado, publica un valor retenido, reinicia el broker y vuelve a ejecutar la prueba 1. Si el valor ha desaparecido, activa la persistencia (en Mosquitto: persistence true y un persistence_location) o asume que los dispositivos deben volver a publicar su estado tras un reinicio del broker. Los brokers gestionados se encargan de esto, pero merece la pena ejecutar la prueba una vez.

Inspecciona mensajes retenidos en tu navegador

Los mensajes retenidos son un contrato pequeño: se guardan al publicar con retain, se entregan una vez por cada suscripción nueva, el siguiente publish retenido los reemplaza y uno vacío los borra. Prueba cada cláusula una vez a mano con el cliente MQTT online o con mosquitto_sub -F '%r', y conserva la suite de pytest para que un cambio de broker nunca rompa en silencio el "último estado conocido" de tus dispositivos.

Preguntas frecuentes

¿Cómo borro un mensaje retenido de MQTT?

Publica un payload de longitud cero con el flag retain en exactamente el mismo topic, por ejemplo mosquitto_pub -t 'mi/topic' -r -n. El broker elimina el mensaje almacenado. No se puede borrar con un wildcard; para limpiar un subárbol entero usa mosquitto_sub --remove-retained con una suscripción con wildcard.

¿Por qué mi suscriptor ve retain=0 en un mensaje publicado como retenido?

Porque ya estaba suscrito. En MQTT 3.1.1 el broker solo marca el flag retain en los mensajes enviados a causa de una suscripción nueva; las entregas en vivo a suscriptores existentes llevan retain=0. En MQTT 5 puedes conservar el flag original con la opción de suscripción Retain As Published.

¿Un publish no retenido borra el mensaje retenido?

No. Un publish normal se entrega a los suscriptores actuales, pero deja intacto el mensaje retenido almacenado. Solo otro publish retenido lo reemplaza, y solo un publish retenido con payload vacío (o la expiración del mensaje en MQTT 5) lo elimina.

Cómo probar los mensajes retenidos (y cómo borrarlos)