Client MQTT TypeScript type-safe : topics typés et payloads validés

Écrire du MQTT en TypeScript vous apporte l'autocomplétion et les vérifications à la compilation, mais seulement si vos topics et vos payloads sont réellement typés. Par défaut, MQTT.js vous remet une chaîne de topic et un Buffer, et tout le reste relève de la devinette. Dans ce tutoriel, vous allez définir un fichier de contrats unique pour les topics, la QoS et les schémas de payload, envelopper MQTT.js dans un petit client typé et valider chaque message à l'exécution, afin que publishers et subscribers ne puissent pas diverger sans que le compilateur ou le validateur vous le signale.

Pourquoi le code MQTT casse en silence

Les brokers MQTT ne savent pas ce que contient un payload et ne s'en soucient pas. Cette souplesse est une force, mais elle signifie que les bugs les plus courants des systèmes MQTT ne déclenchent jamais d'erreur :

Les types statiques détectent les deux premiers à la compilation quand les deux côtés partagent le même code. La validation à l'exécution détecte les deux derniers, car les données provenant des appareils, d'autres équipes ou d'anciens firmwares ne sont jamais passées par votre compilateur. Vous avez besoin des deux.

Mise en place du projet

mkdir mqtt-typescript-demo && cd mqtt-typescript-demo
npm init -y
npm pkg set type=module
npm install mqtt zod
npm install -D typescript tsx @types/node
npx tsc --init --strict --target es2022 --module nodenext --moduleResolution nodenext

tsx exécute directement les fichiers TypeScript, et tsc --noEmit vérifie les types du projet. MQTT.js v5 fournit ses propres définitions de types. zod fournit des schémas qui servent aussi de types TypeScript : vous n'écrivez donc chaque forme de payload qu'une seule fois.

Étape 1 : le fichier de contrats

Créez src/contracts.ts. C'est la source de vérité unique que chaque publisher et chaque subscriber importe :

import { z } from "zod";

export const PREFIX = process.env.TOPIC_PREFIX ?? "connectmq/demo/ts-7f3a";

export const Telemetry = z.object({
  deviceId: z.string(),
  temperature: z.number(),
  humidity: z.number().min(0).max(100),
  ts: z.number().int(),
});

export const Command = z.object({
  action: z.enum(["on", "off", "reboot"]),
  requestId: z.string().uuid(),
});

export const Status = z.enum(["online", "offline"]);

interface Contract<S extends z.ZodType, A extends unknown[]> {
  filter: string;
  topic: (...args: A) => string;
  schema: S;
  qos: 0 | 1 | 2;
  retain: boolean;
}

export const contracts = {
  telemetry: {
    filter: `${PREFIX}/devices/+/telemetry`,
    topic: (deviceId: string) => `${PREFIX}/devices/${deviceId}/telemetry`,
    schema: Telemetry,
    qos: 0,
    retain: false,
  },
  command: {
    filter: `${PREFIX}/devices/+/cmd`,
    topic: (deviceId: string) => `${PREFIX}/devices/${deviceId}/cmd`,
    schema: Command,
    qos: 1,
    retain: false,
  },
  status: {
    filter: `${PREFIX}/devices/+/status`,
    topic: (deviceId: string) => `${PREFIX}/devices/${deviceId}/status`,
    schema: Status,
    qos: 1,
    retain: true,
  },
} as const satisfies Record<string, Contract<z.ZodType, any[]>>;

export type Contracts = typeof contracts;
export type ContractKey = keyof Contracts;
export type PayloadOf<K extends ContractKey> = z.infer<Contracts[K]["schema"]>;
export type TopicArgs<K extends ContractKey> = Parameters<Contracts[K]["topic"]>;

La combinaison as const satisfies remplit deux rôles. satisfies vérifie chaque entrée par rapport à la forme Contract : oublier un schéma ou utiliser une QoS 3 devient une erreur de compilation. as const conserve les types littéraux exacts, si bien que PayloadOf<"command"> se résout en { action: "on" | "off" | "reboot"; requestId: string } plutôt qu'en un objet vague.

Les constructeurs de topics sont des fonctions plutôt que des chaînes à trous : contracts.command.topic(deviceId) est donc vérifié quant au nombre et au type de ses arguments, et plus personne ne concatène de chaînes de topics à la main.

Étape 2 : un client MQTT typé

Créez src/typed-client.ts :

import mqtt, { IClientOptions, MqttClient } from "mqtt";
import { contracts, ContractKey, PayloadOf, TopicArgs } from "./contracts.js";

type Handler<K extends ContractKey> = (payload: PayloadOf<K>, topic: string) => void;

export function topicMatches(filter: string, topic: string): boolean {
  const f = filter.split("/");
  const t = topic.split("/");
  for (let i = 0; i < f.length; i++) {
    if (f[i] === "#") return true;
    if (i >= t.length) return false;
    if (f[i] !== "+" && f[i] !== t[i]) return false;
  }
  return f.length === t.length;
}

export class TypedMqttClient {
  private readonly client: MqttClient;
  private readonly handlers = new Map<ContractKey, Set<Handler<any>>>();

  constructor(
    url: string,
    options: IClientOptions,
    private readonly onInvalid: (topic: string, error: unknown) => void = (topic, error) =>
      console.warn(`Invalid payload on ${topic}:`, error)
  ) {
    this.client = mqtt.connect(url, { protocolVersion: 5, reconnectPeriod: 2000, ...options });

    this.client.on("connect", () => {
      for (const key of this.handlers.keys()) {
        const c = contracts[key];
        this.client.subscribe(c.filter, { qos: c.qos });
      }
    });

    this.client.on("message", (topic, buffer) => this.dispatch(topic, buffer));
    this.client.on("error", (err) => console.error("MQTT error:", err.message));
  }

  async publish<K extends ContractKey>(key: K, args: TopicArgs<K>, payload: PayloadOf<K>): Promise<void> {
    const c = contracts[key];
    const validated = c.schema.parse(payload); // ne jamais envoyer ce que nous rejetterions nous-mêmes
    const topic = (c.topic as (...a: unknown[]) => string)(...args);
    await this.client.publishAsync(topic, JSON.stringify(validated), { qos: c.qos, retain: c.retain });
  }

  subscribe<K extends ContractKey>(key: K, handler: Handler<K>): () => void {
    let set = this.handlers.get(key);
    if (!set) {
      set = new Set();
      this.handlers.set(key, set);
      if (this.client.connected) {
        this.client.subscribe(contracts[key].filter, { qos: contracts[key].qos });
      }
    }
    set.add(handler);
    return () => {
      set!.delete(handler);
      if (set!.size === 0) {
        this.handlers.delete(key);
        this.client.unsubscribe(contracts[key].filter);
      }
    };
  }

  async close(): Promise<void> {
    await this.client.endAsync();
  }

  private dispatch(topic: string, buffer: Buffer): void {
    for (const [key, set] of this.handlers) {
      const c = contracts[key];
      if (!topicMatches(c.filter, topic)) continue;
      let raw: unknown;
      try {
        raw = JSON.parse(buffer.toString());
      } catch (err) {
        this.onInvalid(topic, err);
        continue;
      }
      const result = c.schema.safeParse(raw);
      if (!result.success) {
        this.onInvalid(topic, result.error.issues);
        continue;
      }
      for (const handler of set) handler(result.data, topic);
    }
  }
}

Trois détails rendent ce client robuste :

• Les payloads sortants sont validés eux aussi. TypeScript vérifie déjà les littéraux, mais les valeurs construites à l'exécution (depuis des capteurs, des formulaires ou d'autres services) peuvent encore enfreindre des contraintes comme humidity <= 100. schema.parse lève une exception avant qu'un message invalide ne quitte le processus.
• Les payloads entrants passent par safeParse. Un message mal formé provenant d'un appareil est signalé via onInvalid au lieu de faire planter le subscriber ou de se propager dans la logique métier sous la forme d'un objet mal typé.
• Les abonnements sont rétablis à chaque connexion, avec la QoS de chaque contrat : les reconnexions sont donc invisibles pour le code appelant.

Étape 3 : l'utiliser

Créez src/demo.ts :

import { randomUUID } from "node:crypto";
import { contracts } from "./contracts.js";
import { TypedMqttClient } from "./typed-client.js";

const mq = new TypedMqttClient("mqtts://broker.connectmq.com:8883", {
  clientId: `ts-demo-${randomUUID().slice(0, 8)}`,
});

// payload est entièrement typé : { deviceId: string; temperature: number; humidity: number; ts: number }
mq.subscribe("telemetry", (payload, topic) => {
  console.log(`${payload.deviceId}: ${payload.temperature.toFixed(1)} C (${topic})`);
});

mq.subscribe("command", (cmd) => {
  console.log(`Command ${cmd.action} (${cmd.requestId})`);
});

await mq.publish("status", ["sensor-42"], "online");

const timer = setInterval(() => {
  mq.publish("telemetry", ["sensor-42"], {
    deviceId: "sensor-42",
    temperature: 20 + Math.random() * 5,
    humidity: 40 + Math.random() * 10,
    ts: Date.now(),
  }).catch((err) => console.error("Rejected by schema:", err.message));
}, 2000);

setTimeout(async () => {
  await mq.publish("command", ["sensor-42"], { action: "reboot", requestId: randomUUID() });
}, 5000);

// Ces lignes ne compilent pas, et c'est tout l'intérêt :
// mq.publish("command", ["sensor-42"], { action: "explode", requestId: "x" });
// mq.publish("telemetry", [], { deviceId: "a", temperature: "21", humidity: 1, ts: 0 });

process.on("SIGINT", async () => {
  clearInterval(timer);
  await mq.publish("status", ["sensor-42"], "offline");
  await mq.close();
  process.exit(0);
});

console.log("Telemetry topic example:", contracts.telemetry.topic("sensor-42"));

Exécutez-le et vérifiez les types séparément :

npx tsx src/demo.ts
npx tsc --noEmit

Décommentez les deux lignes « qui ne compilent pas » et tsc signale une valeur d'enum invalide pour action, un argument de topic manquant et une chaîne là où un nombre est attendu. C'est exactement la catégorie de bugs qui, sinon, arrive en production.

Pour voir la validation à l'exécution à l'oeuvre, publiez un payload invalide depuis la ligne de commande sur le serveur MQTT public gratuit, réservé aux tests :

mosquitto_pub -h broker.connectmq.com -p 1883 -i ts-bad-7f3a \
  -t connectmq/demo/ts-7f3a/devices/sensor-9/telemetry -m '{"deviceId":"sensor-9","temperature":"hot"}'

La démo journalise un avertissement Invalid payload listant les champs en échec, et votre handler n'est jamais appelé. L'espace public est lisible par tout le monde : remplacez ts-7f3a par votre propre chaîne aléatoire.

Partager les contrats entre services et équipes

Le fichier de contrats prend toute sa valeur quand chaque composant l'importe :

1. Monorepo : placez contracts.ts dans un package partagé utilisé par les services backend, le tableau de bord React et les outils de test.
2. Dépôts séparés : publiez les contrats sous forme de package npm privé et versionnez-le. Un changement de schéma incompatible devient une montée de version majeure que les consommateurs adoptent volontairement.
3. Appareils hors TypeScript : générez du JSON Schema à partir des schémas zod avec une bibliothèque de conversion et partagez-le avec les équipes firmware, afin que le code ESP32 ou Python puisse valider selon la même définition.
4. Faire évoluer en sécurité : ajoutez d'abord les nouveaux champs en optionnel, déployez les subscribers, puis mettez à jour les publishers. Pour les payloads qui changent complètement de forme, ajoutez un segment de version dans le topic.

Tester vos contrats

Les contrats sont du code : testez-les comme du code. Trois types de tests apportent le plus de valeur pour le moins d'effort :

1. Fixtures de schémas. Conservez un dossier de payloads réels capturés depuis les appareils, y compris d'anciennes versions de firmware, et vérifiez que chacun passe safeParse pour son contrat. Quand quelqu'un durcit un schéma, le test montre immédiatement quels appareils déployés commenceraient à échouer à la validation.
2. Allers-retours de topics. Pour chaque contrat, construisez un topic avec topic() et vérifiez que topicMatches(filter, builtTopic) renvoie true. Cela détecte l'erreur classique consistant à modifier un constructeur sans mettre à jour son filtre.
3. Tests au niveau des types. Des fichiers qui ne doivent pas compiler, vérifiés avec des commentaires // @ts-expect-error, prouvent que les payloads invalides restent rejetés. Si quelqu'un assouplit un type par accident, tsc signale la directive devenue inutile.

Ces trois tests s'exécutent en quelques millisecondes et ne nécessitent aucun broker : ils ont donc leur place dans les vérifications de chaque pull request plutôt que dans un job nocturne. Ajoutez un unique test d'intégration sur le serveur MQTT public gratuit avec un préfixe aléatoire pour vérifier tout le chemin publication, validation et dispatch sur une vraie connexion.

Utiliser un broker privé

Passer du serveur MQTT public gratuit à la production ne change que le constructeur. Sur ConnectMQ, créez un broker dans le portail et transmettez les identifiants générés :

const mq = new TypedMqttClient(`mqtts://${process.env.MQTT_HOST}:8883`, {
  clientId: process.env.MQTT_CLIENT_ID!,   // YOUR_CLIENT_ID
  username: process.env.MQTT_USERNAME!,    // YOUR_USERNAME
  password: process.env.MQTT_PASSWORD!,    // YOUR_PASSWORD
});

Un broker MQTT managé prend en charge TLS et l'exploitation du broker, tandis que vos contrats typés encadrent ce qui y circule. L'explorateur de topics en direct du portail ConnectMQ permet de confirmer rapidement que les topics générés par vos constructeurs correspondent à vos attentes.

Tutoriels associés

Si vous découvrez le client sous-jacent, commencez par MQTT.js avec Node.js. Les mêmes contrats peuvent alimenter les observables du service Angular MQTT. Pour les autres langages et leurs options de typage, consultez le guide des bibliothèques clientes MQTT. La documentation de zod présente des fonctionnalités de schéma comme les refinements et les transforms que vous pouvez ajouter à vos contrats.

Démarrer avec ConnectMQ

Faites tourner vos services MQTT type-safe sur un broker privé en quelques minutes. Inscrivez-vous sur app.connectmq.com, choisissez l'offre Serverless gratuite sans carte bancaire, créez un broker et transmettez les identifiants à TypedMqttClient. L'offre gratuite prend en charge MQTT 5 et la QoS jusqu'à 1, ce qui couvre tous les contrats de ce tutoriel.

Questions fréquentes

MQTT.js prend-il en charge TypeScript ?

Oui. MQTT.js v5 est écrit en TypeScript et fournit ses propres déclarations : les options, les événements et les paquets sont donc typés. Les payloads restent des Buffer, d'où l'utilité d'une couche de contrats comme celle-ci.

Pourquoi valider à l'exécution si TypeScript vérifie déjà les types ?

TypeScript ne vérifie que le code qu'il compile. Les messages provenant des appareils, d'autres services ou d'anciens firmwares arrivent sous forme d'octets non typés : la validation à l'exécution est donc le seul moyen de garantir que l'objet reçu par votre handler correspond à son type déclaré.

Peut-on utiliser autre chose que zod ?

Oui. Tout validateur qui infère des types TypeScript convient, comme Valibot, TypeBox ou ArkType. La structure des contrats reste identique ; seules les définitions de schémas et les appels de parsing changent.

Comment typer des topics avec des wildcards ?

Conservez deux champs par contrat : un filter avec + ou # pour l'abonnement, et un constructeur topic() typé pour la publication. N'utilisez le filtre que côté subscriber et extrayez les paramètres de la chaîne du topic quand un handler a besoin de l'identifiant de l'appareil.