Modèle requête-réponse MQTT avec MQTT 5

Le modèle requête-réponse MQTT permet à un client de poser une question et de recevoir une réponse précise, même si MQTT est en soi un protocole publish/subscribe. MQTT 5 a ajouté deux propriétés, response topic et correlation data, qui rendent ce modèle propre et interopérable. Dans ce guide, vous allez construire un appel de type RPC fonctionnel en Python et Node.js, le tester sur le serveur MQTT public gratuit et apprendre à gérer les timeouts et les échecs.

Pourquoi faire de la requête/réponse sur MQTT

La plupart du trafic MQTT est unidirectionnel : les capteurs publient, les tableaux de bord s'abonnent. Mais les systèmes réels ont aussi besoin de questions avec réponses :

« Quelle version de firmware exécutes-tu ? »
« Lis le registre 40001 et donne-moi sa valeur. »
« La vanne a-t-elle réellement bougé ? »
« Lance un autotest et communique le résultat. »

Vous pourriez ouvrir un endpoint HTTP sur chaque appareil, mais les appareils se trouvent souvent derrière du NAT ou des réseaux cellulaires où les connexions entrantes sont impossibles. Puisque l'appareil maintient déjà une connexion MQTT sortante, faire transiter la requête et la réponse par le broker est plus simple et conserve un modèle de sécurité unique. La comparaison entre MQTT et HTTP explique plus en détail pourquoi ce modèle sortant uniquement est important.

Comment fonctionne le modèle requête-réponse dans MQTT 5

Le modèle utilise deux propriétés PUBLISH définies dans MQTT 5 :

Propriété	Définie par	Rôle
Response Topic	Demandeur	Topic sur lequel le répondeur doit publier la réponse
Correlation Data	Demandeur	Octets opaques que le répondeur renvoie à l'identique

Le déroulement est le suivant :

Le demandeur s'abonne à son propre topic de réponse, par exemple connectmq/demo/k3p8/rpc/replies/<requester-id>.
Il publie la requête sur le topic de requête du répondeur, avec Response Topic défini sur ce topic de réponse et Correlation Data sur un identifiant unique.
Le répondeur reçoit la requête, effectue le travail et publie le résultat sur le Response Topic en recopiant la Correlation Data.
Le demandeur associe la Correlation Data reçue à la requête en attente et la finalise.

Le broker ne suit absolument pas les requêtes. Il se contente de transmettre deux messages ordinaires. Tout ce qui en fait du « RPC » réside dans les clients, ce qui explique pourquoi le modèle passe bien à l'échelle et fonctionne sur n'importe quel broker MQTT 5.

Conception des topics pour la requête/réponse

De bons topics rendent le modèle facile à sécuriser et à déboguer :

Requêtes : <prefix>/rpc/<device-id>/req, pour que chaque appareil ne s'abonne qu'à ses propres requêtes.
Réponses : <prefix>/rpc/replies/<requester-id>, pour que chaque appelant ne reçoive que ses propres réponses.
N'utilisez jamais un topic de réponse partagé entre plusieurs appelants. Chaque appelant recevrait toutes les réponses et devrait écarter la plupart d'entre elles.

Pour une vision plus large de la structuration des noms, consultez les conventions de nommage des topics MQTT.

Répondeur en Python

Ce répondeur joue le rôle d'un appareil. Il écoute les requêtes et renvoie un résultat en utilisant le response topic et la correlation data de la requête. Il utilise paho-mqtt 2.x avec MQTT 5 sur le serveur MQTT public gratuit, réservé aux tests et aux prototypes (pas aux données de production) :

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

BROKER = "broker.connectmq.com"
REQ_TOPIC = "connectmq/demo/k3p8/rpc/pump-07/req"

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(REQ_TOPIC, qos=1)
    print("responder ready")

def on_message(client, userdata, msg):
    props = msg.properties
    response_topic = getattr(props, "ResponseTopic", None)
    correlation = getattr(props, "CorrelationData", None)
    if not response_topic:
        print("request without response topic, ignoring")
        return

    request = json.loads(msg.payload)
    if request.get("method") == "getStatus":
        result = {"ok": True, "rpm": 1480, "running": True}
    else:
        result = {"ok": False, "error": "unknown method"}

    reply_props = Properties(PacketTypes.PUBLISH)
    if correlation is not None:
        reply_props.CorrelationData = correlation
    client.publish(response_topic, json.dumps(result), qos=1, properties=reply_props)

client = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id=f"responder-{uuid.uuid4().hex[:8]}",
    protocol=mqtt.MQTTv5,
)
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=30)
client.loop_forever()

Demandeur Node.js avec timeouts

Le demandeur conserve une table des appels en attente indexée par identifiant de corrélation et rejette tout appel qui n'obtient pas de réponse dans le délai imparti. Il utilise MQTT.js v5 sur TLS :

const mqtt = require("mqtt");
const { randomUUID } = require("crypto");

const clientId = "requester-" + randomUUID().slice(0, 8);
const replyTopic = `connectmq/demo/k3p8/rpc/replies/${clientId}`;
const pending = new Map();

const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId,
  protocolVersion: 5,
});

function call(deviceId, method, params = {}, timeoutMs = 5000) {
  const id = randomUUID();
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      pending.delete(id);
      reject(new Error(`timeout calling ${method} on ${deviceId}`));
    }, timeoutMs);
    pending.set(id, { resolve, timer });
    client.publish(
      `connectmq/demo/k3p8/rpc/${deviceId}/req`,
      JSON.stringify({ method, params }),
      {
        qos: 1,
        properties: {
          responseTopic: replyTopic,
          correlationData: Buffer.from(id),
          messageExpiryInterval: Math.ceil(timeoutMs / 1000),
        },
      }
    );
  });
}

client.on("message", (topic, payload, packet) => {
  const corr = packet.properties && packet.properties.correlationData;
  if (!corr) return;
  const id = corr.toString();
  const entry = pending.get(id);
  if (!entry) return; // réponse tardive ou inconnue
  clearTimeout(entry.timer);
  pending.delete(id);
  entry.resolve(JSON.parse(payload.toString()));
});

client.on("connect", () => {
  client.subscribe(replyTopic, { qos: 1 }, async (err) => {
    if (err) return console.error(err);
    try {
      const status = await call("pump-07", "getStatus");
      console.log("reply:", status);
    } catch (e) {
      console.error(e.message);
    } finally {
      client.end();
    }
  });
});

Démarrez le répondeur Python, puis lancez le demandeur Node.js. Vous devriez voir reply: { ok: true, rpm: 1480, running: true }. Arrêtez le répondeur et relancez le demandeur pour observer le chemin du timeout.

Notez que la requête porte aussi un message expiry égal au timeout. Si l'appareil est hors ligne avec une session persistante, la requête est supprimée dès que l'appelant a abandonné, au lieu d'être exécutée des minutes plus tard. Les détails se trouvent dans notre article sur le message expiry interval MQTT.

Check-list de conception requête/réponse

Abonnez-vous au topic de réponse avant d'envoyer la première requête, sinon vous risquez de manquer une réponse rapide.
Utilisez un identifiant de corrélation unique par requête, comme un UUID.
Implémentez toujours un timeout et nettoyez les entrées en attente.
Définissez un message expiry sur les requêtes pour que les appels périmés ne s'exécutent jamais.
Renvoyez des erreurs structurées (ok: false, code d'erreur) plutôt que de rester silencieux.
Rendez les actions idempotentes autant que possible, car la QoS 1 peut livrer une requête deux fois. Consultez les messages dupliqués MQTT.
Ne mettez ni les requêtes ni les réponses en retain.

Concevoir le payload de réponse

Le transport n'est que la moitié du contrat ; le format de la réponse compte tout autant dès que plusieurs équipes appellent les mêmes appareils. Une enveloppe compacte et cohérente simplifie la vie des appelants :

{"ok": false, "code": "BUSY", "message": "self-test already running", "retryAfter": 10}

Quelques conventions qui tiennent bien en pratique :

Répondez toujours, même en cas d'échec. Un répondeur silencieux est indiscernable d'un répondeur hors ligne, et l'appelant ne peut qu'attendre le timeout.
Utilisez des codes lisibles par la machine comme BUSY, INVALID_PARAMS ou UNKNOWN_METHOD, plus un message lisible par un humain pour les journaux.
Versionnez vos méthodes. getStatus.v2 ou un champ version vous permet de modifier la forme de la réponse sans casser les anciens appelants.
Séparez l'acceptation de l'achèvement. Pour tout ce qui prend plus d'une ou deux secondes, répondez immédiatement avec {"ok": true, "accepted": true, "jobId": "..."} et publiez la progression et le résultat final sur un topic de tâche. Le timeout de l'appelant ne couvre alors que l'acceptation.
Indiquez le type de contenu. Définir la propriété Content Type de MQTT 5 sur application/json pour les requêtes et les réponses aide les outils génériques à les afficher correctement.

Travailler avec des clients MQTT 3.1.1

Si certains appareils ne parlent que MQTT 3.1.1, les propriétés ne sont pas disponibles. La solution de contournement courante consiste à placer les mêmes informations dans le payload, par exemple {"replyTo": "...", "corrId": "...", "method": "getStatus"}. Cela fonctionne, mais toutes les parties doivent s'accorder sur les noms de champs, alors que les propriétés MQTT 5 sont standard et gardent le payload propre. Les flottes hétérogènes font souvent cohabiter les deux jusqu'au retrait des anciens firmwares.

Dépannage

Aucune réponse n'arrive. Vérifiez que le répondeur voit bien le Response Topic. Si l'un des deux côtés se connecte en MQTT 3.1.1, les propriétés sont perdues. Confirmez aussi que le demandeur est abonné exactement au bon topic de réponse.

Les réponses arrivent au mauvais appelant. Votre topic de réponse est probablement partagé. Incluez-y le client ID du demandeur.

La correlation data ne correspond pas. Elle est binaire. Comparez les octets, ou décodez les deux côtés de façon cohérente, comme le font les exemples avec des chaînes UTF-8.

Les appels se bloquent sous la charge. Un répondeur monothread traite une requête à la fois. Les opérations longues doivent répondre avec un statut « accepted » et publier le résultat final plus tard.

Faire du RPC sur ConnectMQ

ConnectMQ prend en charge le response topic, la correlation data et le message expiry de MQTT 5.0 : le code ci-dessus fonctionne donc tel quel avec votre propre broker. Remplacez l'hôte par celui de votre broker et ajoutez username: "YOUR_USERNAME", password: "YOUR_PASSWORD" et clientId: "YOUR_CLIENT_ID" issus du portail. Un broker MQTT managé supprime la partie exploitation pour que vous puissiez vous concentrer sur le contrat de requête. Pour une vue d'ensemble, commencez par le tutoriel MQTT.

Démarrer avec ConnectMQ

Créez un broker gratuit sur app.connectmq.com avec le plan Serverless, sans carte bancaire. Faites tourner le répondeur sur un appareil ou un portable, appelez-le depuis un script, et regardez requêtes et réponses circuler dans l'explorateur de topics en direct du portail ConnectMQ.

Questions fréquentes

Le broker associe-t-il les requêtes aux réponses ?

Non. Le broker transmet deux messages indépendants. C'est le demandeur qui les associe grâce à la correlation data qu'il a choisie.

La correlation data est-elle obligatoire ?

Pas strictement, mais sans elle, un client ayant plusieurs requêtes en cours ne peut pas savoir quelle réponse correspond à quel appel. Définissez-la toujours.

Puis-je utiliser la QoS 0 pour la requête/réponse ?

C'est possible, mais les requêtes perdues n'apparaissent alors que sous forme de timeouts. La QoS 1 avec des handlers idempotents est le choix habituel.

La requête/réponse MQTT remplace-t-elle les API HTTP ?

Pour les commandes et requêtes adressées à des appareils derrière du NAT, elle convient souvent mieux. Pour les transferts de gros fichiers ou les API publiques, HTTP est généralement plus simple.

Modèle requête-réponse MQTT avec response topic et correlation data