Tip Güvenli MQTT TypeScript Client: Tipli Topic'ler ve Doğrulanmış Payload'lar

MQTT kodunu TypeScript ile yazmak size otomatik tamamlama ve derleme zamanı kontrolleri sağlar, ancak yalnızca topic'leriniz ve payload'larınız gerçekten tiplendirilmişse. MQTT.js varsayılan olarak size bir topic dizesi ve bir Buffer verir; sonrası tamamen tahmindir. Bu eğitimde topic'ler, QoS ve payload şemaları için tek bir sözleşme dosyası tanımlayacak, MQTT.js'i küçük ve tipli bir client ile saracak ve her mesajı çalışma zamanında doğrulayacaksınız; böylece publisher'lar ve subscriber'lar derleyici veya doğrulayıcı size haber vermeden birbirinden kopamaz.

MQTT Kodu Neden Sessizce Bozulur?

MQTT broker'ları bir payload'ın içinde ne olduğunu bilmez ve önemsemez. Bu esneklik bir güçtür, ancak MQTT sistemlerindeki en yaygın hataların hiçbir zaman hata fırlatmaması anlamına gelir:

İki taraf da aynı kodu paylaştığında statik tipler ilk ikisini derleme zamanında yakalar. Son ikisini çalışma zamanı doğrulaması yakalar, çünkü cihazlardan, başka ekiplerden veya eski firmware'den gelen veriler hiçbir zaman derleyicinizden geçmemiştir. İkisine de ihtiyacınız var.

Proje Kurulumu

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 TypeScript dosyalarını doğrudan çalıştırır, tsc --noEmit ise projenin tip kontrolünü yapar. MQTT.js v5 kendi tip tanımlarıyla gelir. zod, aynı zamanda TypeScript tipi işlevi gören şemalar sağlar; böylece her payload yapısını yalnızca bir kez yazarsınız.

Adım 1: Sözleşme Dosyası

src/contracts.ts dosyasını oluşturun. Bu, her publisher ve subscriber'ın import ettiği tek doğru kaynaktır:

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

as const satisfies birleşimi iki iş görür. satisfies her girdiyi Contract yapısına göre kontrol eder; böylece bir şemayı unutmak veya QoS 3 kullanmak derleme hatasıdır. as const tam literal tipleri korur; böylece PayloadOf<"command"> belirsiz bir nesne yerine { action: "on" | "off" | "reboot"; requestId: string } olarak çözümlenir.

Topic oluşturucular yer tutuculu dizeler değil fonksiyonlardır; bu da contracts.command.topic(deviceId) çağrısının doğru sayıda ve tipte argüman için kontrol edildiği ve kimsenin bir daha topic dizelerini elle birleştirmeyeceği anlamına gelir.

Adım 2: Tipli Bir MQTT Client'ı

src/typed-client.ts dosyasını oluşturun:

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); // kendimizin reddedeceği bir şeyi asla gönderme
    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);
    }
  }
}

Bunu sağlam kılan üç ayrıntı:

• Giden payload'lar da doğrulanır. TypeScript literal'leri zaten kontrol eder, ancak çalışma zamanında oluşturulan değerler (sensörlerden, formlardan veya diğer servislerden) yine de humidity <= 100 gibi kısıtlamaları ihlal edebilir. schema.parse, hatalı bir mesaj süreçten çıkmadan önce hata fırlatır.
• Gelen payload'lar safeParse kullanır. Bir cihazdan gelen bozuk bir mesaj, subscriber'ı çökertmek veya yanlış tiplenmiş bir nesne olarak iş mantığına akmak yerine onInvalid aracılığıyla raporlanır.
• Abonelikler her bağlantıda geri yüklenir; her sözleşmenin QoS değeri kullanılır, böylece yeniden bağlantılar çağıran kod için görünmez olur.

Adım 3: Kullanın

src/demo.ts dosyasını oluşturun:

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 tamamen tiplidir: { 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);

// Bu satırlar derlenmez; amaç da tam olarak bu:
// 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"));

Çalıştırın ve tip kontrolünü ayrıca yapın:

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

"Derlenmez" olarak işaretlenen iki satırın yorumunu kaldırın; tsc, action için geçersiz bir enum değeri, eksik bir topic argümanı ve sayı gereken yerde bir string olduğunu bildirir. Aksi halde üretime ulaşacak hata sınıfı tam olarak budur.

Çalışma zamanı doğrulamasını iş başında görmek için komut satırından public MQTT test broker'a bozuk bir payload yayınlayın:

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

Demo, hatalı alanları listeleyen bir Invalid payload uyarısı loglar ve handler'ınız hiç çağrılmaz. Public alan herkes tarafından okunabilir; bu yüzden ts-7f3a kısmını kendi rastgele dizenizle değiştirin.

Sözleşmeleri Servisler ve Ekipler Arasında Paylaşmak

Sözleşme dosyası en çok, her bileşen onu import ettiğinde değerlidir:

1. Monorepo: contracts.ts dosyasını backend servisleri, React panosu ve test araçları tarafından kullanılan paylaşılan bir pakete koyun.
2. Ayrı depolar: Sözleşmeleri private bir npm paketi olarak yayınlayın ve sürümleyin. Kırıcı bir şema değişikliği, tüketicilerin bilinçli olarak geçtiği bir major sürüm artışına dönüşür.
3. TypeScript olmayan cihazlar: Bir dönüştürücü kütüphaneyle zod şemalarından JSON Schema üretin ve bunu firmware ekipleriyle paylaşın; böylece ESP32 veya Python kodu aynı tanıma göre doğrulama yapabilir.
4. Güvenle evrimleştirin: Yeni alanları önce isteğe bağlı olarak ekleyin, subscriber'ları dağıtın, ardından publisher'ları güncelleyin. Yapısı tamamen değişen payload'lar için topic'e bir sürüm segmenti ekleyin.

Sözleşmelerinizi Test Etmek

Sözleşmeler koddur, o yüzden onları kod gibi test edin. Üç tür test, en az çabayla en fazla değeri sağlar:

1. Şema fixture'ları. Eski firmware sürümleri dahil cihazlardan yakalanmış gerçek payload'ları içeren bir klasör tutun ve her birinin kendi sözleşmesi için safeParse doğrulamasından geçtiğini test edin. Birisi bir şemayı sıkılaştırdığında test, hangi dağıtılmış cihazların doğrulamada başarısız olmaya başlayacağını hemen gösterir.
2. Topic gidiş-dönüşleri. Her sözleşme için topic() ile bir topic oluşturun ve topicMatches(filter, builtTopic) sonucunun true olduğunu doğrulayın. Bu, bir oluşturucuyu filtresini güncellemeden değiştirme şeklindeki klasik hatayı yakalar.
3. Tip düzeyinde testler. // @ts-expect-error yorumlarıyla kontrol edilen ve derlenmemesi gereken dosyalar, geçersiz payload'ların reddedilmeye devam ettiğini kanıtlar. Birisi bir tipi yanlışlıkla gevşetirse tsc kullanılmayan direktifi işaretler.

Üçü de milisaniyeler içinde çalışır ve broker gerektirmez; bu yüzden gece çalışan bir işe değil, her pull request kontrolüne aittir. Gerçek bir bağlantı üzerinden tam publish, doğrulama ve dağıtım yolunu doğrulamak için rastgele bir önekle public broker'a karşı tek bir entegrasyon testi ekleyin.

Private Bir Broker'a Karşı Çalıştırmak

Public broker'dan üretime geçmek yalnızca constructor'ı değiştirir. ConnectMQ üzerinde portalda bir broker oluşturun ve üretilen kimlik bilgilerini iletin:

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

Yönetilen bir MQTT broker TLS ve broker operasyonlarıyla ilgilenirken tipli sözleşmeleriniz de içinden akan verilerle ilgilenir. ConnectMQ portalının canlı topic gezgini, oluşturucularınızın ürettiği topic'lerin beklentinizle eşleştiğini doğrulamanın hızlı bir yoludur.

İlgili Eğitimler

Alttaki client'a yeniyseniz Node.js'te MQTT.js ile başlayın. Aynı sözleşmeler Angular MQTT servisindeki observable'ların da temeli olabilir. Diğer diller ve tipleme seçenekleri için MQTT client kütüphaneleri rehberine bakın. zod dokümantasyonu, sözleşmelerinize ekleyebileceğiniz refinement ve transform gibi şema özelliklerini ele alır.

ConnectMQ ile Başlayın

Tip güvenli MQTT servislerinizi dakikalar içinde private bir broker üzerinde çalıştırın. app.connectmq.com adresinden kaydolun, kredi kartı gerektirmeyen ücretsiz Serverless planını seçin, bir broker oluşturun ve kimlik bilgilerini TypedMqttClient sınıfına iletin. Ücretsiz plan MQTT 5'i ve QoS 1'e kadar destekler; bu da bu eğitimdeki tüm sözleşmeleri karşılar.

Sık Sorulan Sorular

MQTT.js TypeScript'i destekliyor mu?

Evet. MQTT.js v5 TypeScript ile yazılmıştır ve kendi tanımlarıyla gelir; seçenekler, olaylar ve paketler tiplidir. Payload'lar ise Buffer olarak kalır; bu yazıdaki gibi bir sözleşme katmanının faydalı olmasının nedeni de budur.

TypeScript zaten tipleri kontrol ediyorsa neden çalışma zamanında doğrulama yapmalı?

TypeScript yalnızca derlediği kodu kontrol eder. Cihazlardan, diğer servislerden veya eski firmware'den gelen mesajlar tipsiz baytlar olarak ulaşır; bu yüzden handler'ınızın aldığı nesnenin bildirilen tipiyle eşleştiğini garanti etmenin tek yolu çalışma zamanı doğrulamasıdır.

zod dışında bir şey kullanabilir miyim?

Evet. Valibot, TypeBox veya ArkType gibi TypeScript tiplerini çıkarabilen her doğrulayıcı işe yarar. Sözleşme yapısı aynı kalır; yalnızca şema tanımları ve parse çağrıları değişir.

Wildcard içeren topic'leri nasıl tiplerim?

Sözleşme başına iki alan tutun: abone olmak için + veya # içeren bir filter ve yayın yapmak için tipli bir topic() oluşturucu. Filtreyi yalnızca subscriber tarafında kullanın ve bir handler cihaz ID'sine ihtiyaç duyduğunda parametreleri topic dizesinden çıkarın.