Retained Mesajlar Nasıl Test Edilir (ve Nasıl Temizlenir)

Retained mesajları test etmek için retain bayrağıyla yayın yapın, ardından yepyeni bir subscriber bağlayın ve mesajı hemen retain=1 ile aldığını kontrol edin. Sonra sözleşmenin diğer yarısını test edin: ikinci bir retained publish birincinin üzerine yazar, retain'siz bir publish yazmaz ve boş payload'lu bir retained publish topic'i temizler; böylece sonraki subscriber hiçbir şey almaz.

Bu rehber her durumu mosquitto_pub / mosquitto_sub ile tek tek ele alıyor, bunları otomatik bir pytest paketine dönüştürüyor ve retained davranışını değiştiren MQTT 5 seçeneklerini anlatıyor.

Neleri doğrulamalı

Temel bilgiler MQTT retained mesajlar yazısında anlatılıyor. Paylaşılan bir broker'da başkasının durumunu okumamak için testmqtt/retain-test/... gibi benzersiz bir topic ağacı kullanın — ve sonrasında temizleyin.

Test 1: geç gelen subscriber son değeri alır

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'

Beklenen çıktı: 1 testmqtt/retain-test/door closed. -F '%r %t %p' retain bayrağını, topic'i ve payload'u yazdırır; -C 1 bir mesajdan sonra çıkar, -W 5 beş saniye sonra vazgeçer — böylece komut bir betikte asla takılı kalmaz. Subscriber zaman aşımına uğrarsa mesaj saklanmamıştır: gerçekten -r verdiğinizi, ACL'inizin retained mesaj yayınlamaya izin verdiğini ve broker'ın retain'i destekleyip desteklemediğini kontrol edin (Retain not supported MQTT 5'te 0x9A reason code'udur).

Test 2: canlı teslimatta retain=0 olur

Önce bir subscriber başlatın ve çalışır bırakın, sonra retained bir mesaj yayınlayın:

# 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 önce saklanan değeri 1 ile, ardından yenisini 0 testmqtt/retain-test/door open olarak yazdırır. Bu doğrudur: MQTT 3.1.1'de bayrak subscriber'a "publisher retain'i açtı" demez; "bu, abone olduğun için aldığın saklı durumdur" der. Eski durumu canlı olaylardan ayırmak için msg.retain kullanan kod çalışır — publisher'ın bayrağını görmeyi bekleyen kod çalışmaz. MQTT 5 ile -V mqttv5 --retain-as-published kullanarak abone olursanız Terminal 1 canlı mesaj için de 1 gösterir.

Test 3–5: üzerine yazma, yok sayma, temizleme

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 null (sıfır uzunlukta) bir payload gönderir. Üç ayrıntı sık sık kafa karıştırır:

• Topic birebir eşleşmelidir. testmqtt/retain-test/#'e boş mesaj yayınlamak geçersizdir — publish topic'lerinde wildcard kullanılamaz. Her topic'i tek tek temizleyin ya da aşağıdaki toplu yöntemi kullanın.
• Boş mesaj da teslim edilir. O anda bağlı olan aboneler sıfır uzunlukta bir payload alır. Handler'larınızın b"" ile çökmediğinden emin olun (bir JSON parser çöker).
• Boşluk karakteri boş değildir. -m ' ' bir baytlık retained mesaj saklar. Bir topic "temizlenmiyorsa" bu yaygın bir nedendir — bkz. retained mesaj temizlenmiyor.

Test 6: wildcard'lar ve toplu temizlik

# 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

Wildcard abonelik, eşleşen her topic için bir retained mesaj alır; dolayısıyla bir dağıtımdan sonra geniş bir filtreye abone olmak bir anda binlerce mesaj getirebilir. --remove-retained çalıştırmadan önce bir filtrenin hangi topic'leri kapsadığını topic matcher ile kontrol edin — üretim broker'ında eşleşen her şeyi siler.

pytest ile otomatikleştirme

Aynı sözleşme otomatik bir test paketi olarak (paho-mqtt 2.x). Her test kendi rastgele topic'ini alır ve fixture sonrasında onu temizler:

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

Sabit wait örneği kısa tutar; CI'da bunu ilk mesajda tetiklenen bir event ve kısa bir ek süreyle değiştirin. Yeniden kullanılabilir broker fixture'ları için MQTT için pytest fixture'ları yazısına bakın.

Sonucu değiştiren MQTT 5 seçenekleri

MQTT 5, eksiksiz bir testin kapsaması gereken abonelik seçenekleri ekler (ayrıntılar abonelik seçenekleri yazısında):

• Retain Handling — 0 her subscribe'da retained mesajları gönderir (varsayılan), 1 yalnızca abonelik yeniyse, 2 hiçbir zaman.
• Retain As Published — canlı teslimatlarda publisher'ın retain bayrağını korur (Test 2 bu durumda 1 gösterir).
• Message Expiry Interval — süresi tanımlı retained mesaj, süresi dolunca kaldırılır; bunu mesaj süresi testleri ile sınayın.

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: retained durum yeniden başlatmadan sağ çıkıyor mu?

Broker diske yazmadığı sürece retained mesajlar broker belleğinde yaşar. Kendi barındırdığınız bir broker'da retained bir değer yayınlayın, broker'ı yeniden başlatın ve Test 1'i tekrar çalıştırın. Değer kaybolduysa kalıcılığı açın (Mosquitto için: persistence true ve bir persistence_location) ya da cihazların broker yeniden başladıktan sonra durumlarını yeniden yayınlaması gerektiğini kabul edin. Yönetilen broker'lar bunu sizin yerinize halleder, ama testi bir kez çalıştırmaya yine de değer.

Retained mesajları tarayıcınızda inceleyin

Retained mesajlar küçük bir sözleşmedir: retain ile yapılan publish'te saklanır, her yeni abonelikte bir kez teslim edilir, bir sonraki retained publish ile değiştirilir ve boş bir publish ile temizlenir. Her maddeyi online MQTT client veya mosquitto_sub -F '%r' ile bir kez elle test edin, ardından pytest paketini saklayın ki bir broker değişikliği cihazlarınızın "bilinen son durumunu" asla sessizce bozmasın.

Sıkça sorulan sorular