TypeScript ile Tip Güvenli MQTT: MQTT.js ve Zod

TypeScript ile tip güvenli MQTT iki anlama gelir: derleyici yanlış payload'ı yanlış topic'e publish etmenizi engeller ve gelen her mesaj, kodunuz onu tipli veri olarak kullanmadan önce doğrulanır. MQTT.js 5 kendi tipleriyle gelir, ancak payload'lar Buffer olarak ulaşır ve topic'ler düz string'lerdir. Bu rehber, bu boşluğu tipli bir topic haritası, zod şemalarından türetilen payload tipleri ve MQTT.js client'ı etrafında küçük bir wrapper ile kapatıyor.

Örnekler strict modda MQTT.js 5, zod 4 ve TypeScript 5 kullanır; broker.connectmq.com'a testmqtt/ts/... altındaki topic'lerle bağlanır.

Install

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

Tiplerin işe yaramadığı yer

Tipsiz MQTT.js kodu yanlış olsa bile rahatça derlenir:

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

Yanlış yazılmış bir topic sessizce hiçbir yere gitmez, sayı beklenen yerde string dashboard'a ulaşır ve JSON.parse any döndürür; bu da sonrasındaki her şey için tip kontrolünü kapatır. Burada iki ayrı sorun var: giden mesajlar derleme zamanında, gelen mesajlar ise runtime'da kontrol gerektirir, çünkü broker herhangi bir client'ın gönderdiği her baytı iletir.

Önce şemalar, tipler onlardan

Her payload'ı bir kez zod şeması olarak yazın ve tipi TypeScript'in ondan çıkarmasına izin verin. Böylece runtime kontrolü ile statik tip birbirinden kopamaz:

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

Şemaları firmware simülatörlerinin, backend servislerinin ve testlerin import edebileceği bir pakette tutun. Alan adlandırma ve sürümleme MQTT payload tasarımı yazısında ele alınıyor.

Tipli bir topic haritası

Ardından her topic'i bir kez tanımlayın: adlandırılmış parametreli kalıbı ve şeması. Template literal type'lar parametre adlarını kalıp string'inden çıkarabilir; böylece derleyici telemetry'nin bir deviceId istediğini bilir:

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

Dikkat edilmesi gereken üç ayrıntı:

as const satisfies … kalıp string'lerini literal olarak korur (ParamNames bu sayede çalışır) ve yine de her girdinin bir kalıbı ve bir şeması olduğunu kontrol eder.
topicFor, /, + veya # içeren parametre değerlerini reddeder. Bu kontrol olmadan, kullanıcı girdisinden gelen dev-1/# gibi bir cihaz ID'si publish'i hiç amaçlamadığınız bir şeye dönüştürürdü.
filterFor("telemetry"), {deviceId} yerine + koyar ve testmqtt/ts/devices/+/telemetry subscription filtresini verir. Buradaki kalıplar regex özel karakteri içermez; sizinkiler içeriyorsa escape edin. Filtreleri örnek topic'lere karşı topic matcher ile deneyebilirsiniz.

MqttClient etrafında tipli bir wrapper

Wrapper tek message listener'ının sahibidir. Gelen her topic'i bir anahtara eşler, JSON'u parse eder, ilgili şemayla doğrular ve ancak ondan sonra tipli handler'ları çağırır:

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

Geçersiz mesajlar, event handler içinde exception fırlatmak yerine loglanıp atılır; aksi hâlde hata, nedeninden çok uzakta ortaya çıkardı. Production'da bunları sayın ya da bir dead-letter topic'ine yönlendirin ki hatalı davranan bir firmware sürümünü fark edin. Yaygın hata türleri MQTT JSON parse hataları yazısında listeleniyor.

Kullanım

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

Artık ilk örnekteki hatalar derleme zamanında yakalanır:

Hata	Derleyici mesajı
`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">'`

Bu durumlar için bir tip test dosyasına @ts-expect-error satırları ekleyin ve CI'da tsc --noEmit çalıştırın; biri bir şemayı gevşetirse build kırılır.

MQTT 5 ekstraları

protocolVersion: 5 ile şema sürümünü publish property'lerine koyabilirsiniz; böylece tüketiciler doğrulamadan önce doğru parser'ı seçer:

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

Alıcı taraf bunları message handler'ının üçüncü argümanından okur (packet.properties?.userProperties). Cihazlar TypeScript ile yazılmadıysa MQTT kod üreticisi diğer diller için eşdeğer bağlantı kodunu üretir.

Doğrulamayı test etmek

main.ts'yi npx tsx src/main.ts ile çalıştırın, ardından online MQTT client'ı açın, broker.connectmq.com'a bağlanın ve testmqtt/ts/devices/dev-9/telemetry topic'ine üç mesaj publish edin:

{"deviceId":"dev-9","temperature":19.5,"humidity":40,"ts":1760000000000} — handler mesajı loglar.
{"deviceId":"dev-9","temperature":"hot"} — zod'un issue listesiyle birlikte invalid payload görürsünüz (expected number, received string ve eksik alanlar).
not json — not JSON on testmqtt/ts/devices/dev-9/telemetry görürsünüz.

Aynı üç durum iyi bir unit test olur: TypedMqtt'ye sahte bir client verin, message event'leri yayınlayın ve hangi handler'ların çalıştığını doğrulayın; MQTT.js kodunu Jest ve Vitest ile test etmek yazısında gösterildiği gibi.

Tipli client'a test payload'ları gönderin

Tipli bir topic haritası ve sınırda şema doğrulaması, MQTT'yi "herhangi bir string üzerinde herhangi bir bayt" olmaktan çıkarıp güvenle refactor edebileceğiniz bir arayüze dönüştürür. Topic'leri ve şemaları bir kez tanımlayın, tüm tipleri onlardan türetin, gelen her şeyi doğrulayın ve gerisini derleyiciye bırakın.

Sıkça sorulan sorular

MQTT.js'in TypeScript tipleri var mı?

Evet. MQTT.js 5 TypeScript ile yazılmıştır ve kendi tip bildirimleriyle gelir; MqttClient, IClientOptions ve IClientPublishOptions bir @types paketi olmadan kullanılabilir. Ancak payload'lar Buffer olarak tiplenir; içeriklerini tiplemek size kalır.

Zaten TypeScript kullanıyorsam MQTT payload'larını neden runtime'da doğrulamalıyım?

TypeScript tipleri derleme sırasında kaybolur ve broker her topic'te her baytı kabul eder. Firmware hataları, eski cihaz sürümleri ve elle gönderilen test mesajları tipsiz Buffer'lar olarak gelir. Bir zod şeması her mesajı sınırda bir kez kontrol eder; böylece kodun geri kalanı tipe güvenebilir.

TypeScript wildcard'lı MQTT topic'lerini kontrol edebilir mi?

Template literal type'lar sayesinde oluşturduğunuz topic'leri kontrol edebilir; topic anahtarındaki bir yazım hatası ya da eksik bir parametre derleme hatası olur. Gelen topic'lerin + ve # filtreleriyle eşleştirilmesi yine runtime'da yapılır; bu yüzden wrapper her mesajı tipli bir topic anahtarına geri eşler.

TypeScript ile Tip Güvenli MQTT