MQTT 5 Request/Response Testi: Response Topic ve Correlation Data

MQTT publish/subscribe tabanlıdır, ancak birçok gerçek sistem bir soru ve bir cevaba ihtiyaç duyar: "bu ayarı oku", "bu komutu çalıştır ve işe yarayıp yaramadığını söyle". MQTT 5 request/response bunu iki PUBLISH özelliğiyle standartlaştırır: istek gönderen taraf bir Response Topic (yanıtın gönderileceği yer) ve Correlation Data (yanıtı soruya bağlayan opak bir ID) belirler. Yanıtlayıcı correlation data'yı yanıtına kopyalar ve response topic'e publish eder.

Bu rehber deseni Python ve JavaScript ile uygular, komut satırından tek satırlık bir test gösterir ve gerçekten üretime çıkan hataları yakalayan test senaryolarını listeler: eşleşmeyen ID'ler, eksik timeout'lar, eşzamanlı istekler ve yanıtı engelleyen ACL'ler. Desenin tasarım tarafı için MQTT request/response yazısına bakın.

Akış nasıl işler

Broker burada hiçbir şeyi zorlamaz; özellikleri yalnızca iletir. Bu taşınabilirlik için iyi haber (her MQTT 5 broker'ı çalışır), doğruluk için kötü haberdir: yanıtlayıcınız correlation data'yı kopyalamayı unutursa, isteyen taraf timeout'a düşene kadar hiçbir şey hata vermez.

CONNECT/CONNACK'te iki ilgili özellik daha vardır. İstemci CONNECT içinde Request Response Information ayarlayabilir, broker da CONNACK içinde Response Information ile yanıt verebilir: istemcinin yanıtlar için kullanması gereken bir topic öneki. Destek broker'dan broker'a değişir, bu yüzden çoğu uygulama response topic'i doğrudan kendi client ID'sinden oluşturur. Her durumda response topic bir topic adıdır: + veya # içeremez.

Önce komut satırından test edin

Mosquitto, varsayılan olarak MQTT 5 konuşan bir request/response istemcisi olan mosquitto_rr ile birlikte gelir. Ne geldiğini görmek için bir terminalde sahte bir yanıtlayıcı (özellikleri yazdıran bir subscriber) başlatın:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -t 'testmqtt/rr/req/#' \
  -F 'topic=%t response=%R corr=%D payload=%p'

İkinci bir terminalde mosquitto_rr ile bir istek gönderin. Response topic'e abone olur, isteği publish eder ve bir yanıt bekler:

mosquitto_rr -h broker.connectmq.com -p 1883 \
  -t testmqtt/rr/req/device-7 \
  -e testmqtt/rr/resp/cli-$$ \
  -m '{"op":"get","key":"firmware"}' -W 5

Terminal 1 response topic'i ve correlation data'yı gösterir. Elle yanıt vermek için üçüncü bir terminalden, yazdırılan response topic'e aynı correlation data ile publish edin:

mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -t testmqtt/rr/resp/cli-12345 \
  -D publish correlation-data '<terminal 1de yazdirilan deger>' \
  -m '{"firmware":"2.4.1"}'

mosquitto_rr yanıtı yazdırır ve çıkar; -W 5 ile beş saniye sonra vazgeçer, bu da ilk timeout testinizdir. İki topic'i online MQTT istemcisinde de izleyebilirsiniz: testmqtt/rr/# topic'ine abone olun, her istek ve yanıt yan yana görünür.

Python ile yanıtlayıcı ve isteyen

Aşağıdaki kod MQTT 5 ile paho-mqtt 2.x kullanır. Yanıtlayıcı isteklere abone olur ve correlation data'yı geri gönderir:

import json
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

HOST = "broker.connectmq.com"

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe("testmqtt/rr/req/+", qos=1)

def on_message(client, userdata, msg):
    resp_topic = getattr(msg.properties, "ResponseTopic", None)
    corr = getattr(msg.properties, "CorrelationData", None)
    if not resp_topic:
        print("Response Topic olmayan istek yok sayildi:", msg.topic)
        return
    request = json.loads(msg.payload)
    reply = {"ok": True, "echo": request}
    props = Properties(PacketTypes.PUBLISH)
    if corr is not None:
        props.CorrelationData = corr
    client.publish(resp_topic, json.dumps(reply), qos=1, properties=props)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_message = on_message
client.connect(HOST, 1883)
client.loop_forever()

İsteyen taraf, correlation ID'ye göre indekslenmiş bekleyen istekler tablosu tutar; böylece aynı anda çok sayıda istek yolda olabilir ve yanıtlar herhangi bir sırayla gelebilir:

import json, threading, time, uuid
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

HOST = "broker.connectmq.com"
CLIENT_ID = f"rr-test-{uuid.uuid4().hex[:8]}"
RESP_TOPIC = f"testmqtt/rr/resp/{CLIENT_ID}"
pending = {}  # correlation id -> [Event, yanit]
ready = threading.Event()

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(RESP_TOPIC, qos=1)

def on_subscribe(client, userdata, mid, reason_codes, properties):
    ready.set()

def on_message(client, userdata, msg):
    corr = getattr(msg.properties, "CorrelationData", b"")
    slot = pending.get(corr)
    if slot is None:
        print("beklenmeyen veya gec gelen yanit:", corr)
        return
    slot[1] = json.loads(msg.payload)
    slot[0].set()

def request(client, device, body, timeout=5.0):
    corr = uuid.uuid4().bytes
    slot = pending[corr] = [threading.Event(), None]
    props = Properties(PacketTypes.PUBLISH)
    props.ResponseTopic = RESP_TOPIC
    props.CorrelationData = corr
    props.MessageExpiryInterval = int(timeout)  # bayat istekler atilir
    t0 = time.perf_counter()
    client.publish(f"testmqtt/rr/req/{device}", json.dumps(body), qos=1, properties=props)
    ok = slot[0].wait(timeout)
    pending.pop(corr, None)
    if not ok:
        raise TimeoutError(f"{device} {timeout}s icinde yanit vermedi")
    return slot[1], (time.perf_counter() - t0) * 1000

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID, protocol=mqtt.MQTTv5)
client.on_connect, client.on_subscribe, client.on_message = on_connect, on_subscribe, on_message
client.connect(HOST, 1883)
client.loop_start()
ready.wait(5)

reply, ms = request(client, "device-7", {"op": "get", "key": "firmware"})
print(f"{ms:.1f} ms icinde yanit: {reply}")

İki ayrıntı önemlidir. İsteyen taraf göndermeden önce SUBACK'i bekler; aksi halde hızlı bir yanıtlayıcı abonelik oluşmadan yanıt verebilir ve yanıt kaybolur. Ayrıca istek, timeout'a eşit bir Message Expiry Interval taşır; böylece sonradan çevrimiçi olan bir yanıtlayıcı, çağıranın çoktan vazgeçtiği bir komutu çalıştırmaz (message expiry yazısına bakın).

Aynı desen MQTT.js ile

MQTT.js 5, özellikleri publish seçeneklerinde ve alınan pakette sunar:

import mqtt from "mqtt";
import { randomUUID } from "node:crypto";

const id = `rr-js-${randomUUID().slice(0, 8)}`;
const respTopic = `testmqtt/rr/resp/${id}`;
const client = await mqtt.connectAsync("mqtt://broker.connectmq.com:1883", {
  protocolVersion: 5,
  clientId: id,
});
await client.subscribeAsync(respTopic, { qos: 1 });

const pending = new Map();
client.on("message", (topic, payload, packet) => {
  const corr = packet.properties?.correlationData?.toString();
  pending.get(corr)?.(JSON.parse(payload.toString()));
  pending.delete(corr);
});

function request(device, body, timeoutMs = 5000) {
  const corr = randomUUID();
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      pending.delete(corr);
      reject(new Error(`${device} beklenirken timeout`));
    }, timeoutMs);
    pending.set(corr, (reply) => { clearTimeout(timer); resolve(reply); });
    client.publish(`testmqtt/rr/req/${device}`, JSON.stringify(body), {
      qos: 1,
      properties: { responseTopic: respTopic, correlationData: Buffer.from(corr) },
    });
  });
}

console.log(await request("device-7", { op: "get", key: "firmware" }));
await client.endAsync();

Bunu yukarıdaki Python yanıtlayıcıya karşı çalıştırın: iki kütüphane de yalnızca standart MQTT 5 özelliklerini kullandığı için birlikte çalışır. Kod üreticisi, bağlantı kısmını başka diller için de hazırlayabilir.

Otomatikleştirmeye değer test senaryoları

Bir request/response katmanı, üretimde bozulana kadar basit görünür. Bu senaryolar gerçek hataların çoğunu kapsar:

1. Mutlu yol. Bir istek, bir yanıt; payload ve correlation data eşleşir.
2. Yanıtlayıcı çevrimdışı. İsteyen taraf takılıp kalmamalı, timeout fırlatmalıdır. Yapılandırılan değerde tetiklendiğini ölçün.
3. Geç yanıt. Yanıtlayıcıyı timeout'tan uzun süre uyutun. Geç gelen yanıt yok sayılmalı, sonraki isteğe teslim edilmemelidir.
4. Eşzamanlılık. Rastgele gecikme ekleyen bir yanıtlayıcıyla 100 isteği paralel gönderin; her yanıt doğru çağırana ulaşmalıdır. Sıra dışı yanıtlar normaldir.
5. Eksik özellikler. Response Topic olmadan bir istek gönderin. Yanıtlayıcı çökmemeli; loglayıp isteği atmalıdır.
6. Süresi dolmuş istek. Yanıtlayıcıyı durdurun, 2 saniyelik expiry ile bir istek gönderin, 5 saniye bekleyin ve yanıtlayıcıyı kalıcı oturumla yeniden başlatın. İsteği çalıştırmamalıdır.
7. Yetkilendirme. ACL'ler açıkken yanıtlayıcının response topic'lerde publish, isteyenin de kendi topic'inde subscribe yetkisine ihtiyacı vardır. Reddedilen bir yanıt çoğu zaman hata vermeden düşürülür; sessizce reddedilen ACL yazısına bakın.

Yanıt hiç gelmediğinde

Faydalı bir alışkanlık: yanıtlayıcının, aldığı response topic'i ve correlation data'yı işlemeden önce loglamasını sağlayın. Bir test başarısız olduğunda bu tek log satırı hangi tarafın hatalı olduğunu söyler.

Gidiş-dönüş gecikmesini ölçmek

İsteyen taraf isteğe zaman damgası basıp yanıtı aynı saatle aldığı için request/response, uçtan uca gecikmeyi ölçmek için kolay bir yerdir. request() çağrısını birkaç yüz kez döngüde çalıştırıp p50 ve p99 değerlerini kaydedin. Gidiş-dönüş iki broker atlaması, yanıtlayıcının işlem süresi ve iki ağ yolunu içerir; yanıtlayıcınızın ne kadar süre eklediğini görmek için tek atlamalı bir gecikme testiyle karşılaştırın. Testi QoS 0 ve QoS 1 ile tekrarlayın: ek PUBACK'ler doğrudan sayılara yansır.

İstekleri ve yanıtları canlı izleyin

MQTT 5 üzerinde request/response yalnızca iki özellik gerektirir; ancak ayrıntılar (göndermeden önce abone olmak, timeout'lar, expiry, benzersiz correlation ID'ler ve doğru ACL'ler) güvenilir olup olmadığını belirler. Yukarıdaki betikleri bir regresyon paketi olarak saklayın ve her broker veya izin değişikliğinde çalıştırın.

Sıkça sorulan sorular