Tester le request/response MQTT 5 avec Response Topic et Correlation Data

MQTT est un protocole publish/subscribe, mais beaucoup de systèmes réels ont besoin d'une question et d'une réponse : « lis ce paramètre », « exécute cette commande et dis-moi si elle a réussi ». Le request/response MQTT 5 standardise cela avec deux propriétés du PUBLISH : le demandeur définit un Response Topic (où envoyer la réponse) et des Correlation Data (un ID opaque qui relie la réponse à la question). Le répondeur recopie la correlation data dans sa réponse et publie sur le response topic.

Ce guide implémente le pattern en Python et en JavaScript, montre un test en une ligne depuis le terminal et liste les cas de test qui attrapent les bugs réellement livrés en production : IDs qui ne correspondent pas, timeouts absents, requêtes concurrentes et ACL qui bloquent la réponse. Pour la partie conception, voir request/response MQTT.

Fonctionnement du flux

Le broker n'impose rien ici : il se contente de relayer les propriétés. C'est une bonne nouvelle pour la portabilité (n'importe quel broker MQTT 5 convient) et une mauvaise pour la justesse : si votre répondeur oublie de recopier la correlation data, rien ne se plaint jusqu'à ce que le demandeur tombe en timeout.

Il existe deux propriétés liées dans CONNECT/CONNACK. Un client peut définir Request Response Information dans le CONNECT, et le broker peut répondre avec Response Information dans le CONNACK — un préfixe de topic que le client devrait utiliser pour les réponses. Le support varie selon les brokers, si bien que la plupart des applications construisent simplement le response topic à partir de leur client ID. Dans tous les cas, le response topic est un nom de topic : il ne doit contenir ni + ni #.

Testez d'abord en ligne de commande

Mosquitto fournit mosquitto_rr, un client request/response qui parle MQTT 5 par défaut. Démarrez un faux répondeur dans un terminal — un subscriber qui affiche les propriétés — pour voir ce qui arrive :

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

Dans un deuxième terminal, envoyez une requête avec mosquitto_rr. Il s'abonne au response topic, publie la requête et attend une réponse :

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

Le terminal 1 affiche le response topic et la correlation data. Pour répondre à la main, publiez depuis un troisième terminal sur le response topic affiché avec la même correlation data :

mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -t testmqtt/rr/resp/cli-12345 \
  -D publish correlation-data '<valeur affichée dans le terminal 1>' \
  -m '{"firmware":"2.4.1"}'

mosquitto_rr affiche la réponse et se termine ; avec -W 5, il abandonne au bout de cinq secondes, ce qui constitue votre premier test de timeout. Vous pouvez aussi suivre les deux topics dans le client MQTT en ligne : abonnez-vous à testmqtt/rr/# et chaque requête et réponse apparaît côte à côte.

Un répondeur et un demandeur en Python

Le code suivant utilise paho-mqtt 2.x avec MQTT 5. Le répondeur s'abonne aux requêtes et renvoie la correlation data :

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("requête sans Response Topic, ignorée :", 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()

Le demandeur tient une table des requêtes en attente indexée par correlation ID, de sorte que de nombreuses requêtes peuvent être en vol en même temps et que les réponses peuvent arriver dans n'importe quel ordre :

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, réponse]
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("réponse inattendue ou tardive :", 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)  # les requêtes périmées sont abandonnées
    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"pas de réponse de {device} en {timeout}s")
    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"réponse en {ms:.1f} ms : {reply}")

Deux détails comptent. Le demandeur attend le SUBACK avant d'envoyer — sinon un répondeur rapide peut répondre avant que l'abonnement n'existe et la réponse est perdue. Et la requête porte un Message Expiry Interval égal au timeout, pour qu'un répondeur qui se connecte plus tard n'exécute pas une commande que l'appelant a déjà abandonnée (voir message expiry).

Le même pattern avec MQTT.js

MQTT.js 5 expose les propriétés dans les options de publication et sur le paquet reçu :

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(`timeout en attendant ${device}`));
    }, 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();

Exécutez-le face au répondeur Python ci-dessus : les deux bibliothèques interopèrent car elles n'utilisent que des propriétés MQTT 5 standard. Le générateur de code produit la partie connexion pour d'autres langages.

Cas de test à automatiser

Une couche request/response paraît triviale jusqu'à ce qu'elle échoue en production. Ces cas couvrent la plupart des bugs réels :

1. Cas nominal. Une requête, une réponse ; payload et correlation data correspondent.
2. Répondeur hors ligne. Le demandeur doit lever un timeout, pas rester bloqué. Vérifiez qu'il se déclenche à la valeur configurée.
3. Réponse tardive. Faites dormir le répondeur plus longtemps que le timeout. La réponse tardive doit être ignorée, pas livrée à la requête suivante.
4. Concurrence. Lancez 100 requêtes en parallèle avec un répondeur qui ajoute un délai aléatoire ; chaque réponse doit arriver au bon appelant. Les réponses dans le désordre sont normales.
5. Propriétés manquantes. Envoyez une requête sans Response Topic. Le répondeur doit la journaliser et l'ignorer, pas planter.
6. Requête expirée. Arrêtez le répondeur, envoyez une requête avec un expiry de 2 secondes, attendez 5 secondes, puis redémarrez le répondeur avec une session persistante. Il ne doit pas exécuter la requête.
7. Autorisation. Avec des ACL actives, le répondeur a besoin du droit publish sur les response topics et le demandeur du droit subscribe sur le sien. Une réponse refusée est souvent supprimée sans erreur — voir ACL refusée silencieusement.

Quand la réponse n'arrive jamais

Une bonne habitude : faites journaliser par le répondeur le response topic et la correlation data reçus avant le traitement. Quand un test échoue, cette seule ligne de log vous dit quel côté est en tort.

Mesurer la latence aller-retour

Comme le demandeur horodate la requête et reçoit la réponse avec la même horloge, le request/response est un endroit idéal pour mesurer la latence de bout en bout. Appelez request() en boucle quelques centaines de fois et relevez p50 et p99. L'aller-retour inclut deux passages par le broker, le temps de traitement du répondeur et deux trajets réseau : comparez-le donc à un test de latence à un seul saut pour voir combien de temps votre répondeur ajoute. Répétez en QoS 0 et QoS 1 : les PUBACK supplémentaires se voient directement dans les chiffres.

Voir requêtes et réponses en direct

Le request/response sur MQTT 5 ne demande que deux propriétés, mais les détails — s'abonner avant d'envoyer, timeouts, expiry, IDs de corrélation uniques et ACL correctes — décident de sa fiabilité. Conservez les scripts ci-dessus comme suite de régression et exécutez-les à chaque changement de broker ou de permissions.

Questions fréquentes