MQTT 5 Message Expiry Interval Nasıl Test Edilir

MQTT 5 Message Expiry Interval, broker'a bir mesajın ne kadar süre işe yarar kalacağını söyler. Bu süre dolduğunda mesaj hâlâ teslim edilmeyi bekliyorsa broker, bayat veriyi teslim etmek yerine mesajı atmak zorundadır. Bunu test etmek için kalıcı bir subscriber çevrimdışıyken farklı expiry değerleriyle mesajlar yayınlar, bekler, yeniden bağlanır ve hangi mesajların hangi kalan süreyle geldiğini kontrol edersiniz.

Bu rehber, bu testi mosquitto_pub / mosquitto_sub, kendi sonucunu doğrulayan bir Python betiği ve retained mesajlar için ayrı bir testle adım adım gösterir.

Mesaj süresi dolması gerçekte nasıl çalışır

Publisher, bir PUBLISH paketine Message Expiry Interval özelliğini (saniye cinsinden) ekler. Bundan sonrası broker'ın sorumluluğundadır:

Test açısından kilit nokta: expiry ancak bir mesaj beklediğinde görünür hâle gelir. Sürekli bağlı bir subscriber ile yapılan test hiçbir şey kanıtlamaz. Çevrimdışı bir oturuma ya da retained bir mesaja ihtiyacınız var. Teori ve tasarım önerileri için MQTT mesaj süresi dolması yazısına bakın.

Ön koşullar

• MQTT 5 destekleyen bir broker. Bir MQTT 5 bağlantısının CONNACK paketine bakın ya da önce genel MQTT 5 özellik testlerini çalıştırın.
• Çevrimdışıyken oturumunu koruyan bir subscriber: sabit client ID, Clean Start false ve sıfırdan büyük bir Session Expiry Interval. Oturumun kendisinin süresi dolarsa kuyruğu da onunla gider ve testiniz yanlış şeyi ölçer. Mekanizmayı kalıcı oturumlar yazısı açıklar.
• Hem abonelikte hem yayında QoS 1. Pek çok broker, çevrimdışı istemciler için QoS 0 mesajları hiç kuyruğa almaz.

Örneklerde broker.connectmq.com ve testmqtt/expiry/ altındaki topic'ler kullanılıyor. Testinizin başka kullanıcılarla çakışmaması için rastgele bir sonek ekleyin.

Mosquitto CLI ile hızlı manuel test

Adım 1: kalıcı bir oturum oluşturup abone olun, ardından subscriber'ı Ctrl+C ile durdurun.

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 Clean Start'ı kapatır, -x 600 oturumu bağlantı kesildikten sonra 10 dakika tutar ve %E alınan her mesajın expiry aralığını yazdırır.

Adım 2: subscriber çevrimdışıyken üç mesaj yayınlayın:

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'

Adım 3: yaklaşık 30 saniye bekleyin, sonra 1. adımdaki komutu aynen tekrar çalıştırın. Doğru çalışan bir broker long (expiry değeri 300'ün belirgin şekilde altında) ve none mesajlarını teslim eder, short mesajını ise asla teslim etmez.

short gelirse broker kuyruktaki mesajlarda expiry'yi yok sayıyordur. Hiçbir şey gelmiyorsa oturumunuz korunmamıştır: önce expiry'ye değil, oturuma bakın.

Kendi sonucunu doğrulayan bir Python testi

Manuel test bir kez için yeterlidir. Regresyonları yakalamak için sonucu doğrulayan bir betik kullanın. Bu betik paho-mqtt 2.x ve MQTT 5 özelliklerini kullanır:

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)

Başarılı bir çalıştırma session present: True ve 105'in biraz altında bir long değeri yazdırır. Bir saniyelik farklar normaldir, çünkü broker'lar tam saniyeye yuvarlar.

session present: False ile birlikte ABORT çıktısı, broker'ın oturumu attığı ya da hiç saklamadığı anlamına gelir; bu durumda expiry değerlendirilemez. Paylaşımlı genel test endpoint'leri çoğu zaman çevrimdışı oturumları hiç tutmaz, bu yüzden bu testi kendi broker instance'ınıza karşı çalıştırın. Aksi hâlde önce oturum ayarlarını kontrol edin: oturum süresi dolması testi bu durumu ele alıyor.

Retained mesajların süresinin dolmasını test etmek

Retained mesajlar, expiry'nin önemli olduğu ikinci yerdir ve en çok unutulanıdır. 30 saniyelik expiry'ye sahip retained bir "cihaz çevrimiçi" durumu, cihaz onu yenilemeyi bıraktığında kendiliğinden kaybolur.

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

İlk çıktıda %r, retained bir mesaj için 1 yazdırır. İkinci çalıştırmada hiçbir şey yazdırılmamalıdır. Eski değer hâlâ görünüyorsa broker retained mesajları süresi dolduktan sonra da tutuyordur. Bazı broker'lar süresi dolan retained mesajları tembel biçimde ya da bir zamanlayıcıyla siler, bu yüzden pay bırakarak test edin. Retained davranışının genelini retained mesajları test etmek yazısında bulabilirsiniz.

Bunu çevrimiçi MQTT istemcisinde canlı olarak da izleyebilirsiniz: expiry penceresinden önce ve sonra durum topic'ine abone olun ve karşılaştırın.

Süresi dolan mesajları tarayıcıda izleyin

Sık görülen sonuçlar ve anlamları

• Kısa ömürlü mesaj, süresi dolduktan sonra teslim ediliyor. Broker özelliği yok sayıyor ya da oturum bir MQTT 3.1.1 kod yolundan geçiyor. Publisher ve subscriber'ın protokol sürümü 5 ile bağlandığını doğrulayın.
• İletilen expiry orijinal değere eşit. Broker bekleme süresini düşmüyor. Bu durum, mesajı yeniden ileten bridge veya gateway gibi sonraki adımların mantığını bozar.
• Expiry ayarlandığı hâlde alınan mesajlarda özellik yok. Aradaki bir bridge veya proxy MQTT 3.1.1 konuşuyor ve MQTT 5 özelliklerini siliyor olabilir.
• Hiçbir şey teslim edilmiyor. session_present değerini, QoS seviyelerini (QoS 0 çoğu zaman kuyruğa alınmaz) ve ACL'leri kontrol edin. Broker'ın reddettiği bir abonelik hiçbir şeyi kuyruğa almaz. Çevrimiçi reason code aracı SUBACK sonuçlarını çözmenize yardımcı olur.
• Mesajların süresi beklenenden geç doluyor. Broker'lar expiry'yi teslim anında ya da periyodik bir taramayla kontrol eder. Bir iki saniyelik fark sorun değildir, dakikalar ise sorundur.

Expiry testleri için kontrol listesi

1. Publisher ve subscriber'ı MQTT 5 ile bağlayın ve CONNACK'i loglayın.
2. Sabit bir client ID, Clean Start false ve testten uzun bir oturum süresi kullanın.
3. QoS 1 ile en az üç expiry değeriyle yayın yapın: beklemeden kısa, beklemeden uzun ve hiç olmayan.
4. Subscriber'ı kısa expiry süresinden daha uzun süre çevrimdışı tutun. Yeniden bağlanın ve neyin geldiğini doğrulayın.
5. İletilen aralığın yaklaşık bekleme süresi kadar azaldığını kontrol edin.
6. Aynısını retained bir mesajla ve süre dolduktan sonra gelen yeni bir subscriber ile tekrarlayın.
7. Temizlik yapın: oturumu Session Expiry 0 ile sonlandırın ve retained topic'leri boş payload ile temizleyin.

Mesaj süresi dolması küçük ama etkisi büyük bir özelliktir: uyanan bir cihazın işe yarar komutlar mı yoksa bir yığın bayat komut mu alacağına o karar verir. Yukarıdaki betiği test paketinize ekleyin ve her broker güncellemesinden ya da taşımasından sonra çalıştırın.

Sıkça sorulan sorular