MQTT fortement typé en TypeScript avec MQTT.js et Zod

MQTT fortement typé en TypeScript signifie deux choses : le compilateur vous empêche de publier le mauvais payload sur le mauvais topic, et chaque message reçu est validé avant que votre code ne le traite comme une donnée typée. MQTT.js 5 fournit ses propres types, mais les payloads arrivent sous forme de Buffer et les topics sont de simples chaînes. Ce guide comble ce manque avec une map de topics typée, des types de payload inférés de schémas zod et un petit wrapper autour du client MQTT.js.

Les exemples utilisent MQTT.js 5, zod 4 et TypeScript 5 en mode strict, contre broker.connectmq.com avec des topics sous testmqtt/ts/....

Install

npm install mqtt zod
npm install -D typescript @types/node

Là où les types lâchent

Du code MQTT.js non typé compile sans broncher, même quand il est faux :

untyped.ts

client.publish("testmqtt/ts/devices/dev-1/telemtry", JSON.stringify({ temp: "21" }));

client.on("message", (topic, payload) => {
  const data = JSON.parse(payload.toString()); // any
  console.log(data.temperature.toFixed(1));     // crashes at runtime
});

Un topic mal orthographié part dans le vide sans prévenir, une chaîne à la place d'un nombre atterrit dans le dashboard, et JSON.parse renvoie any, ce qui désactive la vérification de types pour tout ce qui suit. Ce sont deux problèmes distincts : les messages sortants ont besoin de vérifications à la compilation, les messages entrants de vérifications à l'exécution, car le broker relaie tous les octets envoyés par n'importe quel client.

Les schémas d'abord, les types en découlent

Écrivez chaque payload une seule fois sous forme de schéma zod et laissez TypeScript en inférer le type. Ainsi, la vérification à l'exécution et le type statique ne peuvent pas diverger :

src/schemas.ts

import { z } from "zod";

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(["reboot", "identify", "set-interval"]),
  intervalSec: z.number().int().positive().optional(),
});

export const Status = z.object({
  online: z.boolean(),
  firmware: z.string().optional(),
});

export type Telemetry = z.infer<typeof Telemetry>;

Placez les schémas dans un paquet que les simulateurs de firmware, les services backend et les tests peuvent tous importer. Le nommage des champs et le versioning sont traités dans la conception de payloads MQTT.

Une map de topics typée

Décrivez ensuite chaque topic une fois : son motif avec des paramètres nommés et son schéma. Les template literal types savent extraire les noms de paramètres de la chaîne du motif, si bien que le compilateur sait que telemetry exige un deviceId :

src/topics.ts

import { z } from "zod";
import { Command, Status, Telemetry } from "./schemas";

type ParamNames<T extends string> =
  T extends `${string}{${infer P}}${infer Rest}` ? P | ParamNames<Rest> : never;
type Params<T extends string> = Record<ParamNames<T>, string>;

export const topics = {
  telemetry: { pattern: "testmqtt/ts/devices/{deviceId}/telemetry", schema: Telemetry },
  command: { pattern: "testmqtt/ts/devices/{deviceId}/command", schema: Command },
  status: { pattern: "testmqtt/ts/devices/{deviceId}/status", schema: Status },
} as const satisfies Record<string, { pattern: string; schema: z.ZodType }>;

export type TopicKey = keyof typeof topics;
export type PayloadOf<K extends TopicKey> = z.infer<(typeof topics)[K]["schema"]>;
export type ParamsOf<K extends TopicKey> = Params<(typeof topics)[K]["pattern"]>;

export function topicFor<K extends TopicKey>(key: K, params: ParamsOf<K>): string {
  return topics[key].pattern.replace(/\{(\w+)\}/g, (_, name: string) => {
    const value = (params as Record<string, string>)[name];
    if (!value || /[+#\/]/.test(value)) {
      throw new Error(`invalid topic parameter ${name}=${value}`);
    }
    return value;
  });
}

export function filterFor(key: TopicKey): string {
  return topics[key].pattern.replace(/\{\w+\}/g, "+");
}

export function matchTopic(topic: string) {
  for (const key of Object.keys(topics) as TopicKey[]) {
    const names: string[] = [];
    const source = topics[key].pattern.replace(/\{(\w+)\}/g, (_, name: string) => {
      names.push(name);
      return "([^/]+)";
    });
    const m = new RegExp(`^${source}$`).exec(topic);
    if (m) {
      const params = Object.fromEntries(names.map((n, i) => [n, m[i + 1]]));
      return { key, params };
    }
  }
  return null;
}

Trois détails méritent l'attention :

as const satisfies … conserve les chaînes littérales des motifs (pour que ParamNames fonctionne) tout en vérifiant que chaque entrée possède un motif et un schéma.
topicFor rejette les valeurs de paramètre contenant /, + ou #. Sans ce contrôle, un ID d'appareil comme dev-1/# issu d'une saisie utilisateur transformerait un publish en tout autre chose que prévu.
filterFor("telemetry") remplace {deviceId} par + et produit le filtre d'abonnement testmqtt/ts/devices/+/telemetry. Ces motifs ne contiennent aucun caractère spécial de regex ; échappez-les si les vôtres en ont. Vous pouvez tester des filtres sur des topics d'exemple avec le topic matcher.

Un wrapper typé autour de MqttClient

Le wrapper possède l'unique listener message. Il ramène chaque topic reçu à une clé, parse le JSON, le valide avec le schéma correspondant et n'appelle qu'ensuite les handlers typés :

src/typed-mqtt.ts

import type { IClientPublishOptions, MqttClient } from "mqtt";
import type { z } from "zod";
import {
  filterFor, matchTopic, topicFor, topics,
  type ParamsOf, type PayloadOf, type TopicKey,
} from "./topics";

type AnyHandler = (payload: unknown, params: Record<string, string>) => void;

export class TypedMqtt {
  private readonly client: MqttClient;
  private readonly handlers = new Map<TopicKey, AnyHandler[]>();

  constructor(client: MqttClient) {
    this.client = client;
    client.on("message", (topic, buf) => this.dispatch(topic, buf));
  }

  async publish<K extends TopicKey>(
    key: K,
    params: ParamsOf<K>,
    payload: PayloadOf<K>,
    opts: IClientPublishOptions = { qos: 1 },
  ): Promise<void> {
    const schema: z.ZodType = topics[key].schema;
    const data = schema.parse(payload); // strips unknown fields, catches casts
    await this.client.publishAsync(topicFor(key, params), JSON.stringify(data), opts);
  }

  async on<K extends TopicKey>(
    key: K,
    handler: (payload: PayloadOf<K>, params: ParamsOf<K>) => void,
    qos: 0 | 1 | 2 = 1,
  ): Promise<void> {
    const list = this.handlers.get(key) ?? [];
    list.push(handler as unknown as AnyHandler); // the one place types are erased
    this.handlers.set(key, list);
    await this.client.subscribeAsync(filterFor(key), { qos });
  }

  private dispatch(topic: string, buf: Buffer): void {
    const hit = matchTopic(topic);
    if (!hit) return;

    let json: unknown;
    try {
      json = JSON.parse(buf.toString("utf8"));
    } catch {
      console.warn(`[mqtt] not JSON on ${topic}`);
      return;
    }

    const schema: z.ZodType = topics[hit.key].schema;
    const result = schema.safeParse(json);
    if (!result.success) {
      console.warn(`[mqtt] invalid payload on ${topic}`, result.error.issues);
      return;
    }
    for (const h of this.handlers.get(hit.key) ?? []) h(result.data, hit.params);
  }
}

Les messages invalides sont journalisés puis ignorés, au lieu de lever une exception dans l'event handler, où elle apparaîtrait loin de sa cause. En production, comptez-les ou renvoyez-les vers un topic de dead-letter pour repérer une version de firmware défaillante. Les cas d'échec courants sont décrits dans les erreurs de parsing JSON MQTT.

Utilisation

src/main.ts

import mqtt from "mqtt";
import { TypedMqtt } from "./typed-mqtt";

const client = await mqtt.connectAsync("mqtts://broker.connectmq.com:8883", {
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
});
const bus = new TypedMqtt(client);

await bus.on("telemetry", (t, { deviceId }) => {
  // t is { deviceId: string; temperature: number; humidity: number; ts: number }
  console.log(deviceId, t.temperature.toFixed(1), `${t.humidity}%`);
});

await bus.publish(
  "telemetry",
  { deviceId: "dev-1" },
  { deviceId: "dev-1", temperature: 21.4, humidity: 48, ts: Date.now() },
);

await bus.publish("command", { deviceId: "dev-1" }, { action: "set-interval", intervalSec: 30 });

Les erreurs du premier exemple échouent désormais à la compilation :

Erreur	Message du compilateur
`bus.publish("telemtry", …)`	`Argument of type '"telemtry"' is not assignable to parameter of type '"telemetry" \| "command" \| "status"'`
`{ temperature: "21" }`	`Type 'string' is not assignable to type 'number'`
`{ action: "format" }`	`Type '"format"' is not assignable to type '"reboot" \| "identify" \| "set-interval"'`
`bus.publish("status", {}, …)`	`Property 'deviceId' is missing in type '{}' but required in type 'ParamsOf<"status">'`

Ajoutez des lignes @ts-expect-error pour ces cas dans un fichier de tests de types et lancez tsc --noEmit en CI : si quelqu'un assouplit un schéma, le build échoue.

Les bonus de MQTT 5

Avec protocolVersion: 5, vous pouvez placer la version du schéma dans les propriétés du publish, pour que les consommateurs choisissent le bon parser avant de valider :

Publish with MQTT 5 properties

await client.publishAsync("testmqtt/ts/devices/dev-1/telemetry", JSON.stringify(data), {
  qos: 1,
  properties: {
    contentType: "application/json",
    userProperties: { schema: "telemetry/v2" },
  },
});

Le récepteur les lit dans le troisième argument du handler message (packet.properties?.userProperties). Le générateur de code MQTT produit l'extrait de connexion équivalent pour d'autres langages si les appareils ne sont pas écrits en TypeScript.

Tester la validation

Lancez main.ts avec npx tsx src/main.ts, ouvrez le client MQTT en ligne, connectez-vous à broker.connectmq.com et publiez trois messages sur testmqtt/ts/devices/dev-9/telemetry :

{"deviceId":"dev-9","temperature":19.5,"humidity":40,"ts":1760000000000} — le handler l'affiche.
{"deviceId":"dev-9","temperature":"hot"} — vous voyez invalid payload avec la liste des issues de zod (expected number, received string et les champs manquants).
not json — vous voyez not JSON on testmqtt/ts/devices/dev-9/telemetry.

Ces trois cas font un bon test unitaire : donnez à TypedMqtt un faux client, émettez des événements message et vérifiez quels handlers ont été appelés, comme montré dans tester MQTT.js avec Jest et Vitest.

Envoyer des payloads de test au client typé

Une map de topics typée et une validation de schéma à la frontière font passer MQTT de « n'importe quels octets sur n'importe quelle chaîne » à une interface que vous pouvez refactoriser en confiance. Définissez topics et schémas une fois, dérivez-en tous les types, validez tout ce qui arrive et laissez le compilateur attraper le reste.

Questions fréquentes

MQTT.js fournit-il des types TypeScript ?

Oui. MQTT.js 5 est écrit en TypeScript et embarque ses propres déclarations : MqttClient, IClientOptions et IClientPublishOptions sont disponibles sans paquet @types. Les payloads sont toutefois typés comme Buffer ; typer leur contenu reste à votre charge.

Pourquoi valider les payloads MQTT à l'exécution si j'utilise déjà TypeScript ?

Les types TypeScript disparaissent à la compilation et le broker accepte n'importe quels octets sur n'importe quel topic. Bugs de firmware, anciennes versions d'appareils et messages de test manuels arrivent sous forme de Buffers non typés. Un schéma zod vérifie chaque message une fois à la frontière, pour que le reste du code puisse faire confiance au type.

TypeScript peut-il vérifier les topics MQTT avec wildcards ?

Il vérifie les topics que vous construisez, grâce aux template literal types : une faute de frappe dans une clé de topic ou un paramètre manquant devient une erreur de compilation. La correspondance des topics reçus avec les filtres + et # se fait toujours à l'exécution, c'est pourquoi le wrapper ramène chaque message à une clé de topic typée.

MQTT fortement typé en TypeScript