Testar mensagens retidas MQTT e como apagá-las

Para testar mensagens retidas, publique com a flag retain, conecte em seguida um assinante totalmente novo e verifique se ele recebe a mensagem imediatamente com retain=1. Depois teste a outra metade do contrato: um segundo publish retido sobrescreve o primeiro, um publish não retido não, e um publish retido com payload vazio apaga o topic, de forma que o próximo assinante não recebe nada.

Este guia percorre cada caso com mosquitto_pub / mosquitto_sub, transforma tudo em uma suíte automatizada com pytest e cobre as opções do MQTT 5 que mudam o comportamento das mensagens retidas.

O que verificar

#	Teste	Resultado esperado
1	Publish retido e depois um assinante novo	Mensagem entregue na hora, flag retain 1
2	Assinante já conectado recebe um publish retido	Entregue, flag retain 0 (padrão do MQTT 3.1.1)
3	Segundo publish retido no mesmo topic	Só o valor mais recente fica armazenado
4	Publish não retido no mesmo topic	Valor armazenado inalterado
5	Publish retido com payload vazio	Topic limpo; novos assinantes não recebem nada
6	Assinatura com wildcard sobre topics retidos	Uma mensagem retida por topic correspondente
7	Reinício do broker	Mensagens retidas só sobrevivem se o broker as persistir

O básico está explicado em mensagens retidas do MQTT. Use uma árvore de topics única, como testmqtt/retain-test/..., para não ler o estado de outra pessoa em um broker compartilhado — e limpe tudo no final.

Teste 1: um assinante tardio recebe o ú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'

Saída esperada: 1 testmqtt/retain-test/door closed. -F '%r %t %p' imprime a flag retain, o topic e o payload; -C 1 encerra após uma mensagem e -W 5 desiste depois de cinco segundos, então o comando nunca trava em um script. Se o assinante estourar o tempo, a mensagem não foi armazenada: confira se você realmente passou -r, se a sua ACL permite publicar mensagens retidas e se o broker suporta retain (Retain not supported é o reason code 0x9A no MQTT 5).

Teste 2: entrega ao vivo tem retain=0

Inicie primeiro um assinante e deixe-o rodando; depois publique uma mensagem retida:

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

O terminal 1 imprime primeiro o valor armazenado com 1 e depois o novo com 0 testmqtt/retain-test/door open. Está correto: no MQTT 3.1.1 a flag diz ao assinante "isto é estado armazenado que você recebeu porque assinou", e não "o publisher ativou retain". Código que usa msg.retain para separar estado antigo de eventos ao vivo funciona — código que espera ver a flag do publisher, não. Com MQTT 5, assine com -V mqttv5 --retain-as-published e o terminal 1 mostra 1 também para a mensagem ao vivo.

Testes 3–5: sobrescrever, ignorar, apagar

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 envia um payload nulo (de tamanho zero). Três detalhes costumam pegar as pessoas:

O topic precisa ser exatamente o mesmo. Publicar uma mensagem vazia em testmqtt/retain-test/# é inválido — wildcards não são permitidos em topics de publicação. Apague cada topic ou use o método em massa abaixo.
A mensagem vazia também é entregue. Assinantes conectados naquele momento recebem um payload de tamanho zero. Garanta que seus handlers não quebrem com b"" (um parser JSON vai quebrar).
Um espaço não é vazio. -m ' ' armazena uma mensagem retida de um byte. Se um topic "não apaga", essa é uma causa comum — veja mensagem retida que não apaga.

Teste 6: wildcards e limpeza em massa

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

Uma assinatura com wildcard recebe uma mensagem retida por topic correspondente, então assinar um filtro amplo depois de um deploy pode entregar milhares de mensagens de uma vez. Confira quais topics um filtro cobre com o topic matcher antes de rodar --remove-retained — em um broker de produção ele apaga tudo o que corresponder.

Automatizando com pytest

O mesmo contrato como suíte automatizada (paho-mqtt 2.x). Cada teste recebe seu próprio topic aleatório e o fixture o limpa no 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) == []

O wait fixo deixa o exemplo curto; no CI, troque-o por um evento disparado na primeira mensagem mais uma pequena margem. Para fixtures de broker reutilizáveis veja fixtures do pytest para MQTT.

Opções do MQTT 5 que mudam o resultado

O MQTT 5 adiciona opções de assinatura que um teste completo deve cobrir (detalhes em opções de assinatura):

Retain Handling — 0 envia mensagens retidas a cada subscribe (padrão), 1 só quando a assinatura é nova, 2 nunca.
Retain As Published — mantém a flag retain do publisher nas entregas ao vivo (o teste 2 passa a mostrar 1).
Message Expiry Interval — uma mensagem retida com expiração é removida quando expira; teste isso com os testes de expiração de mensagens.

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

Teste 7: o estado retido sobrevive a um reinício?

Mensagens retidas ficam na memória do broker, a menos que ele as persista em disco. Em um broker self-hosted, publique um valor retido, reinicie o broker e rode o teste 1 de novo. Se o valor sumiu, ative a persistência (no Mosquitto: persistence true e um persistence_location) ou aceite que os dispositivos precisam republicar o estado após um reinício do broker. Brokers gerenciados cuidam disso para você, mas vale a pena rodar o teste uma vez.

Inspecione mensagens retidas no navegador

Mensagens retidas são um contrato pequeno — armazenadas no publish com retain, entregues uma vez por nova assinatura, substituídas pelo próximo publish retido e apagadas por um vazio. Teste cada cláusula uma vez à mão com o cliente MQTT online ou mosquitto_sub -F '%r' e mantenha a suíte pytest para que uma mudança de broker nunca quebre em silêncio o "último estado conhecido" dos seus dispositivos.

Perguntas frequentes

Como apagar uma mensagem retida do MQTT?

Publique um payload de tamanho zero com a flag retain exatamente no mesmo topic, por exemplo mosquitto_pub -t 'meu/topic' -r -n. O broker apaga a mensagem armazenada. Não dá para apagar com wildcard; para limpar uma subárvore inteira use mosquitto_sub --remove-retained com uma assinatura com wildcard.

Por que meu assinante vê retain=0 em uma mensagem publicada como retida?

Porque ele já estava inscrito. No MQTT 3.1.1 o broker só marca a flag retain nas mensagens enviadas por causa de uma nova assinatura; entregas ao vivo para assinantes existentes têm retain=0. No MQTT 5 você pode manter a flag original com a opção de assinatura Retain As Published.

Um publish não retido apaga a mensagem retida?

Não. Um publish normal é entregue aos assinantes atuais, mas não mexe na mensagem retida armazenada. Só outro publish retido a substitui, e só um publish retido com payload vazio (ou a expiração da mensagem no MQTT 5) a remove.

Como testar mensagens retidas (e como apagá-las)