Typsicheres MQTT in TypeScript mit MQTT.js und Zod

Typsicheres MQTT in TypeScript bedeutet zweierlei: Der Compiler verhindert, dass du den falschen Payload auf das falsche Topic publizierst, und jede eingehende Nachricht wird validiert, bevor dein Code sie als typisierte Daten behandelt. MQTT.js 5 bringt eigene Typen mit, aber Payloads kommen als Buffer an und Topics sind einfache Strings. Diese Anleitung schließt die Lücke mit einer typisierten Topic-Map, aus Zod-Schemas abgeleiteten Payload-Typen und einem kleinen Wrapper um den MQTT.js-Client.

Die Beispiele nutzen MQTT.js 5, Zod 4 und TypeScript 5 im Strict-Modus gegen broker.connectmq.com mit Topics unter testmqtt/ts/....

Install

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

Wo die Typen versagen

Untypisierter MQTT.js-Code kompiliert problemlos, auch wenn er falsch ist:

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
});

Ein falsch geschriebenes Topic geht stillschweigend ins Leere, ein String statt einer Zahl landet im Dashboard, und JSON.parse liefert any, was die Typprüfung für alles Nachfolgende abschaltet. Es sind zwei getrennte Probleme: ausgehende Nachrichten brauchen Prüfungen zur Compile-Zeit, eingehende Nachrichten Prüfungen zur Laufzeit, denn der Broker leitet jede Bytefolge weiter, die irgendein Client sendet.

Erst Schemas, dann abgeleitete Typen

Schreibe jeden Payload einmal als Zod-Schema und lass TypeScript den Typ daraus ableiten. So können Laufzeitprüfung und statischer Typ nicht auseinanderlaufen:

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>;

Lege die Schemas in ein Paket, das Firmware-Simulatoren, Backend-Services und Tests gleichermaßen importieren können. Feldbenennung und Versionierung behandelt MQTT-Payload-Design.

Eine typisierte Topic-Map

Als Nächstes beschreibst du jedes Topic einmal: sein Muster mit benannten Parametern und sein Schema. Template Literal Types können die Parameternamen aus dem Muster-String herausziehen, sodass der Compiler weiß, dass telemetry eine deviceId braucht:

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;
}

Drei Details sind wichtig:

as const satisfies … behält die literalen Muster-Strings (damit ParamNames funktioniert) und prüft trotzdem, dass jeder Eintrag ein Muster und ein Schema hat.
topicFor lehnt Parameterwerte mit /, + oder # ab. Ohne diese Prüfung würde eine Geräte-ID wie dev-1/# aus einer Benutzereingabe einen Publish in etwas verwandeln, das du nicht beabsichtigt hast.
filterFor("telemetry") ersetzt {deviceId} durch + und ergibt den Subscription-Filter testmqtt/ts/devices/+/telemetry. Die Muster hier enthalten keine Regex-Sonderzeichen; escape sie, falls deine welche enthalten. Filter gegen Beispiel-Topics prüfst du mit dem Topic Matcher.

Ein typisierter Wrapper um MqttClient

Der Wrapper besitzt den einzigen message-Listener. Er ordnet jedes eingehende Topic einem Schlüssel zu, parst JSON, validiert mit dem passenden Schema und ruft erst dann die typisierten Handler auf:

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);
  }
}

Ungültige Nachrichten werden geloggt und verworfen, statt im Event-Handler eine Exception zu werfen, die weit entfernt von ihrer Ursache auftauchen würde. In Produktion solltest du sie zählen oder an ein Dead-Letter-Topic weiterleiten, damit dir eine fehlerhafte Firmware-Version auffällt. Die typischen Fehlerbilder findest du unter MQTT-JSON-Parse-Fehler.

Verwendung

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 });

Jetzt scheitern die Fehler aus dem ersten Beispiel schon beim Kompilieren:

Fehler	Compiler-Meldung
`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">'`

Lege für diese Fälle @ts-expect-error-Zeilen in einer Typ-Testdatei an und führe tsc --noEmit in CI aus; lockert jemand ein Schema, schlägt der Build fehl.

MQTT-5-Extras

Mit protocolVersion: 5 kannst du die Schema-Version in die Publish-Properties schreiben, damit Konsumenten vor der Validierung den richtigen Parser wählen:

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" },
  },
});

Die Empfängerseite liest sie aus dem dritten Argument des message-Handlers (packet.properties?.userProperties). Der MQTT-Code-Generator erzeugt passende Verbindungs-Snippets für andere Sprachen, falls die Geräte nicht in TypeScript geschrieben sind.

Die Validierung testen

Starte main.ts mit npx tsx src/main.ts, öffne den Online-MQTT-Client, verbinde dich mit broker.connectmq.com und publiziere drei Nachrichten auf testmqtt/ts/devices/dev-9/telemetry:

{"deviceId":"dev-9","temperature":19.5,"humidity":40,"ts":1760000000000} — der Handler loggt sie.
{"deviceId":"dev-9","temperature":"hot"} — du siehst invalid payload mit Zods Issue-Liste (expected number, received string und die fehlenden Felder).
not json — du siehst not JSON on testmqtt/ts/devices/dev-9/telemetry.

Dieselben drei Fälle ergeben einen guten Unit-Test: Gib TypedMqtt einen Fake-Client, löse message-Events aus und prüfe, welche Handler gelaufen sind, wie in MQTT.js mit Jest und Vitest testen gezeigt.

Test-Payloads an den typisierten Client senden

Eine typisierte Topic-Map plus Schema-Validierung an der Grenze macht aus MQTT statt „beliebige Bytes auf beliebigen Strings“ eine Schnittstelle, die du mit Zuversicht refaktorieren kannst. Definiere Topics und Schemas einmal, leite alle Typen daraus ab, validiere alles, was ankommt, und lass den Compiler den Rest erledigen.

Häufig gestellte Fragen

Hat MQTT.js TypeScript-Typen?

Ja. MQTT.js 5 ist in TypeScript geschrieben und liefert eigene Deklarationen mit, MqttClient, IClientOptions und IClientPublishOptions sind also ohne @types-Paket verfügbar. Payloads sind allerdings als Buffer typisiert; deren Inhalt zu typisieren ist deine Aufgabe.

Warum MQTT-Payloads zur Laufzeit validieren, wenn ich schon TypeScript nutze?

TypeScript-Typen verschwinden beim Kompilieren, und der Broker akzeptiert beliebige Bytes auf jedem Topic. Firmware-Bugs, alte Geräteversionen und manuelle Testnachrichten kommen als untypisierte Buffer an. Ein Zod-Schema prüft jede Nachricht einmal an der Grenze, damit der restliche Code dem Typ vertrauen kann.

Kann TypeScript MQTT-Wildcard-Topics prüfen?

Es prüft die Topics, die du baust, mit Template Literal Types, sodass ein Tippfehler im Topic-Schlüssel oder ein fehlender Parameter ein Compile-Fehler ist. Das Abgleichen eingehender Topics mit +- und #-Filtern passiert weiterhin zur Laufzeit, deshalb ordnet der Wrapper jede Nachricht einem typisierten Topic-Schlüssel zu.

Typsicheres MQTT in TypeScript