Typsicherer MQTT TypeScript Client: Typisierte Topics und validierte Payloads

MQTT in TypeScript zu schreiben bringt Autovervollständigung und Prüfungen zur Compile-Zeit – aber nur, wenn Ihre Topics und Payloads auch tatsächlich typisiert sind. Von Haus aus liefert Ihnen MQTT.js einen Topic-String und einen Buffer, und alles danach ist Rätselraten. In diesem Tutorial definieren Sie eine Vertragsdatei für Topics, QoS und Payload-Schemas, verpacken MQTT.js in einen kleinen typisierten Client und validieren jede Nachricht zur Laufzeit. So können Publisher und Subscriber nicht auseinanderdriften, ohne dass Compiler oder Validator Alarm schlagen.

Warum MQTT-Code unbemerkt kaputtgeht

MQTT-Broker wissen nicht, was in einem Payload steht, und es ist ihnen auch egal. Diese Flexibilität ist eine Stärke, bedeutet aber, dass die häufigsten Fehler in MQTT-Systemen nie einen Fehler auslösen:

Statische Typen fangen die ersten beiden zur Compile-Zeit ab, wenn beide Seiten denselben Code nutzen. Laufzeitvalidierung fängt die letzten beiden ab, denn Daten von Geräten, anderen Teams oder älterer Firmware sind nie durch Ihren Compiler gelaufen. Sie brauchen beides.

Projekt-Setup

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 führt TypeScript-Dateien direkt aus, und tsc --noEmit prüft die Typen des Projekts. MQTT.js v5 bringt eigene Typdefinitionen mit. zod liefert Schemas, die zugleich als TypeScript-Typen dienen, sodass Sie jede Payload-Form nur einmal schreiben.

Schritt 1: Die Vertragsdatei

Erstellen Sie src/contracts.ts. Das ist die einzige Quelle der Wahrheit, die jeder Publisher und Subscriber importiert:

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

Die Kombination as const satisfies erledigt zwei Aufgaben. satisfies prüft jeden Eintrag gegen die Form Contract, sodass ein vergessenes Schema oder QoS 3 zum Compile-Fehler wird. as const behält die exakten Literaltypen bei, sodass PayloadOf<"command"> zu { action: "on" | "off" | "reboot"; requestId: string } aufgelöst wird statt zu einem vagen Objekt.

Topic-Builder sind Funktionen statt Strings mit Platzhaltern. Damit wird contracts.command.topic(deviceId) auf die richtige Anzahl und den richtigen Typ der Argumente geprüft, und niemand setzt Topic-Strings je wieder von Hand zusammen.

Schritt 2: Ein typisierter MQTT-Client

Erstellen Sie 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); // nie senden, was wir selbst ablehnen würden
    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);
    }
  }
}

Drei Details machen ihn robust:

• Auch ausgehende Payloads werden validiert. TypeScript prüft Literale bereits, aber zur Laufzeit erzeugte Werte (aus Sensoren, Formularen oder anderen Diensten) können Bedingungen wie humidity <= 100 dennoch verletzen. schema.parse wirft einen Fehler, bevor eine fehlerhafte Nachricht den Prozess verlässt.
• Eingehende Payloads verwenden safeParse. Eine fehlerhafte Nachricht eines Geräts wird über onInvalid gemeldet, statt den Subscriber abstürzen zu lassen oder als falsch typisiertes Objekt in die Geschäftslogik zu gelangen.
• Subscriptions werden bei jedem Connect wiederhergestellt, jeweils mit der QoS des Vertrags, sodass Reconnects für den aufrufenden Code unsichtbar sind.

Schritt 3: Verwenden

Erstellen Sie 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 ist vollständig typisiert: { 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);

// Diese Zeilen kompilieren nicht – genau darum geht es:
// 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"));

Führen Sie es aus und prüfen Sie die Typen separat:

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

Kommentieren Sie die beiden nicht kompilierenden Zeilen ein, und tsc meldet einen ungültigen Enum-Wert für action, ein fehlendes Topic-Argument und einen String, wo eine Zahl verlangt ist. Genau diese Klasse von Fehlern gelangt sonst in die Produktion.

Um die Laufzeitvalidierung in Aktion zu sehen, publizieren Sie über die Kommandozeile einen fehlerhaften Payload an den öffentlichen MQTT-Broker:

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"}'

Die Demo protokolliert eine Invalid payload-Warnung mit den fehlerhaften Feldern, und Ihr Handler wird nie aufgerufen. Der öffentliche MQTT-Broker ist für alle lesbar und nur für Tests gedacht, nicht für Produktionsdaten; ändern Sie ts-7f3a also in eine eigene Zufallszeichenfolge.

Verträge über Dienste und Teams hinweg teilen

Die Vertragsdatei ist am wertvollsten, wenn jede Komponente sie importiert:

1. Monorepo: Legen Sie contracts.ts in ein gemeinsames Paket, das von Backend-Diensten, dem React-Dashboard und Testwerkzeugen genutzt wird.
2. Getrennte Repositories: Veröffentlichen Sie die Verträge als privates npm-Paket und versionieren Sie es. Eine inkompatible Schemaänderung wird so zu einem Major-Versionssprung, für den sich Konsumenten bewusst entscheiden.
3. Geräte ohne TypeScript: Erzeugen Sie mit einer Konverter-Bibliothek JSON Schema aus den zod-Schemas und geben Sie es an Firmware-Teams weiter, damit ESP32- oder Python-Code gegen dieselbe Definition validieren kann.
4. Sicher weiterentwickeln: Fügen Sie neue Felder zunächst optional hinzu, deployen Sie die Subscriber und aktualisieren Sie dann die Publisher. Bei Payloads, die ihre Form komplett ändern, ergänzen Sie ein Versionssegment im Topic.

In industriellen Projekten mit einem Unified Namespace (UNS), dessen Topics der ISA-95-Hierarchie folgen (etwa acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur), wird eine solche Vertragsdatei zur maschinenlesbaren Beschreibung des Namespace, auf die sich MES-Integration, Dashboards und Edge-Dienste gleichermaßen stützen.

Ihre Verträge testen

Verträge sind Code, also testen Sie sie wie Code. Drei Arten von Tests bringen den größten Nutzen bei geringstem Aufwand:

1. Schema-Fixtures. Halten Sie einen Ordner mit echten, von Geräten mitgeschnittenen Payloads vor, auch von alten Firmware-Versionen, und prüfen Sie, dass jeder davon safeParse für seinen Vertrag besteht. Verschärft jemand ein Schema, zeigt der Test sofort, welche ausgerollten Geräte an der Validierung scheitern würden.
2. Topic-Roundtrips. Bauen Sie für jeden Vertrag ein Topic mit topic() und prüfen Sie, dass topicMatches(filter, builtTopic) true ist. Das fängt den klassischen Fehler ab, einen Builder zu ändern, ohne seinen Filter anzupassen.
3. Tests auf Typebene. Dateien, die nicht kompilieren sollen, abgesichert mit // @ts-expect-error-Kommentaren, belegen, dass ungültige Payloads abgelehnt bleiben. Lockert jemand versehentlich einen Typ, meldet tsc die ungenutzte Direktive.

Alle drei laufen in Millisekunden und brauchen keinen Broker – sie gehören also in jede Pull-Request-Prüfung statt in einen nächtlichen Job. Ergänzen Sie einen einzigen Integrationstest gegen den öffentlichen MQTT-Broker mit zufälligem Präfix, um den gesamten Pfad aus Publish, Validierung und Dispatch über eine echte Verbindung zu verifizieren.

Betrieb mit einem privaten Broker

Der Wechsel vom öffentlichen MQTT-Broker in die Produktion ändert nur den Konstruktor. Erstellen Sie auf ConnectMQ im Portal einen Broker und übergeben Sie die generierten Zugangsdaten:

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

Ein verwalteter MQTT-Broker kümmert sich um TLS und den Broker-Betrieb, während Ihre typisierten Verträge sich darum kümmern, was durch ihn fließt. Der Live-Topic-Explorer im ConnectMQ-Portal ist ein schneller Weg, um zu bestätigen, dass die von Ihren Buildern erzeugten Topics Ihren Erwartungen entsprechen.

Verwandte Tutorials

Sind Sie neu beim zugrunde liegenden Client, beginnen Sie mit MQTT.js in Node.js. Dieselben Verträge können auch die Observables im Angular-MQTT-Service absichern. Andere Sprachen und deren Typisierungsoptionen behandelt der Leitfaden zu MQTT-Client-Bibliotheken. Die zod-Dokumentation beschreibt Schema-Features wie Refinements und Transforms, die Sie Ihren Verträgen hinzufügen können.

Starten Sie mit ConnectMQ

Betreiben Sie Ihre typsicheren MQTT-Dienste in wenigen Minuten auf einem privaten Broker. Registrieren Sie sich unter app.connectmq.com, wählen Sie den kostenlosen Serverless-Plan ohne Kreditkarte, erstellen Sie einen Broker und übergeben Sie die Zugangsdaten an TypedMqttClient. Der kostenlose Plan unterstützt MQTT 5 und QoS bis 1 und deckt damit jeden Vertrag in diesem Tutorial ab.

Häufig gestellte Fragen

Unterstützt MQTT.js TypeScript?

Ja. MQTT.js v5 ist in TypeScript geschrieben und bringt eigene Deklarationen mit, sodass Optionen, Events und Pakete typisiert sind. Payloads bleiben Buffer – genau deshalb ist eine Vertragsschicht wie diese nützlich.

Warum zur Laufzeit validieren, wenn TypeScript die Typen schon prüft?

TypeScript prüft nur Code, den es kompiliert. Nachrichten von Geräten, anderen Diensten oder älterer Firmware kommen als untypisierte Bytes an, daher ist Laufzeitvalidierung der einzige Weg zu garantieren, dass das Objekt, das Ihr Handler erhält, seinem deklarierten Typ entspricht.

Kann ich etwas anderes als zod verwenden?

Ja. Jeder Validator, der TypeScript-Typen ableitet, funktioniert, etwa Valibot, TypeBox oder ArkType. Die Vertragsstruktur bleibt gleich; nur die Schemadefinitionen und die Parse-Aufrufe ändern sich.

Wie typisiere ich Topics mit Wildcards?

Führen Sie pro Vertrag zwei Felder: einen filter mit + oder # zum Abonnieren und einen typisierten topic()-Builder zum Publizieren. Verwenden Sie den Filter nur auf der Subscriber-Seite und extrahieren Sie Parameter aus dem Topic-String, wenn ein Handler die Geräte-ID braucht.