NestJS ile MQTT mikroservisleri: transport, pattern'ler ve request/response

NestJS, @nestjs/microservices içinde bir MQTT transport'u sunar; böylece bir Nest servisi cihaz telemetrisini tüketebilir ve request/response çağrılarını bir broker üzerinden, diğer transport'larda kullandığınız decorator'larla yanıtlayabilir. Transport.MQTT'yi yapılandırır, @EventPattern (fire-and-forget) veya @MessagePattern (request/response) ile handler'lar yazar ve diğer servisleri bir ClientProxy ile çağırırsınız.

Bu eğitimde küçük bir telemetri servisi kuruyoruz, Nest'in ağa tam olarak ne yazdığını görüyoruz ve her parçayı gerçek bir broker ile test ediyoruz.

Kurulum ve başlatma

MQTT transport'u arka planda MQTT.js kullanır; bunu sizin kurmanız gerekir:

npm i @nestjs/microservices mqtt

Bir Nest uygulaması saf bir mikroservis ya da MQTT'yi de dinleyen hibrit bir HTTP uygulaması olabilir. Mesaj handler'larının yanında genellikle health check'ler ve bir REST API de istendiği için en yaygın kurulum hibrit olandır:

import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);

  app.connectMicroservice<MicroserviceOptions>({
    transport: Transport.MQTT,
    options: {
      url: 'mqtts://broker.connectmq.com:8883',
      username: process.env.MQTT_USER,
      password: process.env.MQTT_PASS,
      clientId: `telemetry-svc-${process.pid}`,
      subscribeOptions: { qos: 1 },
    },
  });

  await app.startAllMicroservices();
  await app.listen(3000);
}
bootstrap();

options nesnesi MQTT.js'e aktarılır; dolayısıyla MQTT.js'in kabul ettiği her şey (protocolVersion: 5, keepalive, reconnectPeriod, ca) burada çalışır. Her instance'a benzersiz bir client ID verin; aynı ID'ye sahip iki replica birbirini döngü halinde broker'dan düşürür. Bir şema seçmek için client ID oluşturucu işinize yarar.

HTTP'siz saf bir mikroservis için NestFactory.createMicroservice<MicroserviceOptions>(AppModule, { transport: Transport.MQTT, options }) kullanıp app.listen() çağırın.

Event handler'ları: cihazlardan telemetri

@EventPattern bir topic'e abone olur ve her mesaj için handler'ınızı çalıştırır. Pattern'de MQTT wildcard'larına izin verilir; MqttContext de size gerçek topic'i ve ham paketi verir:

import { Controller, Logger } from '@nestjs/common';
import { Ctx, EventPattern, MqttContext, Payload } from '@nestjs/microservices';

interface Reading { t: number; h?: number }

@Controller()
export class TelemetryController {
  private readonly log = new Logger(TelemetryController.name);

  @EventPattern('testmqtt/nestjs/sensors/+/temperature')
  onTemperature(@Payload() reading: Reading, @Ctx() ctx: MqttContext) {
    const deviceId = ctx.getTopic().split('/')[3];
    const { qos, retain } = ctx.getPacket();
    this.log.log(`${deviceId}: ${reading.t} °C (qos ${qos}, retain ${retain})`);
  }
}

Cihazların Nest hakkında hiçbir şey bilmesi gerekmez. Bir payload Nest'in { "pattern": ..., "data": ... } zarfında değilse Nest topic'i pattern olarak, ayrıştırılmış JSON'u (JSON değilse ham string'i) da payload olarak kullanır. testmqtt/nestjs/sensors/dev-42/temperature topic'ine {"t": 21.4} publish eden bir sensör doğrudan onTemperature'a düşer.

Pattern'inizin cihazlarınızın publish ettiği topic'lerle gerçekten eşleştiğini topic matcher ile kontrol edin. Eşleşmeyen bir pattern sessizce başarısız olur: mesaj broker'a ulaşır ama kimse onu işlemez.

@MessagePattern ile request/response

@MessagePattern bir topic'i uzak bir prosedüre dönüştürür. Handler'ın döndürdüğü şey (bir değer, Promise veya Observable) çağırana geri gönderilir:

import { Controller } from '@nestjs/common';
import { MessagePattern, Payload } from '@nestjs/microservices';

@Controller()
export class MathController {
  @MessagePattern('testmqtt/nestjs/sum')
  sum(@Payload() numbers: number[]): number {
    return numbers.reduce((a, b) => a + b, 0);
  }
}

Ağ üzerinde Nest, MQTT 5'in Response Topic özelliği yerine basit bir kural kullanır:

Çağıran taraf <pattern>/reply topic'ine abone olur ve yanıtları id ile eşleştirir. Bunun üretimde iki önemli sonucu vardır: servis hesabının yanıt topic'inde publish ve subscribe yetkisine ihtiyacı vardır ve aynı pattern'i çağıran herkes tüm yanıtları alır (sadece diğer id'leri yok sayar). Yüksek hacimli RPC için pattern'leri spesifik tutun. MQTT'deki genel tasarım için MQTT üzerinden request/response yazısına bakın.

ClientProxy ile bir servisi çağırmak

Çağrıyı yapan servisin modülüne bir client kaydedin:

import { Module } from '@nestjs/common';
import { ClientsModule, Transport } from '@nestjs/microservices';
import { MathController } from './math.controller';
import { TelemetryController } from './telemetry.controller';
import { ReportService } from './report.service';

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'MQTT_CLIENT',
        transport: Transport.MQTT,
        options: {
          url: 'mqtts://broker.connectmq.com:8883',
          username: process.env.MQTT_USER,
          password: process.env.MQTT_PASS,
        },
      },
    ]),
  ],
  controllers: [MathController, TelemetryController],
  providers: [ReportService],
})
export class AppModule {}

Ardından inject edin ve request/response için send(), event'ler için emit() çağırın:

import { Inject, Injectable } from '@nestjs/common';
import { ClientProxy, MqttRecordBuilder } from '@nestjs/microservices';
import { firstValueFrom, timeout } from 'rxjs';

@Injectable()
export class ReportService {
  constructor(@Inject('MQTT_CLIENT') private readonly mqtt: ClientProxy) {}

  async total(values: number[]): Promise<number> {
    return firstValueFrom(
      this.mqtt.send<number, number[]>('testmqtt/nestjs/sum', values).pipe(timeout(5000)),
    );
  }

  announce(status: string) {
    const record = new MqttRecordBuilder({ status, at: Date.now() })
      .setQoS(1)
      .setRetain(true)
      .build();
    return this.mqtt.emit('testmqtt/nestjs/status', record);
  }
}

Üç ayrıntı sık sık sorun çıkarır:

• send() soğuk bir Observable döndürür. Biri abone olana kadar hiçbir şey publish edilmez; firstValueFrom bunu sizin yerinize yapar.
• Varsayılan bir timeout yoktur. Hiçbir handler yanıt vermezse çağrı sonsuza kadar bekler. Her zaman bir timeout() ekleyin.
• MqttRecordBuilder mesaj bazında seçenekleri belirler: QoS, retain ve protocolVersion: 5 ile setProperties() üzerinden user property'ler. Kullanmazsanız mesajlar client varsayılanlarıyla gider.

Ölçekleme: birden fazla replica

Aynı servisin üç replica'sını çalıştırdığınızda her biri aynı topic'lere abone olur; yani her event üç kez işlenir. Veritabanına yazdığınız telemetri için bu genellikle yanlıştır. MQTT 5 shared subscription'ları bunu çözer: broker her mesajı gruptaki yalnızca bir üyeye teslim eder. Güncel Nest sürümleri pattern'lerde $share/<group>/ önekini kabul eder:

@EventPattern('$share/telemetry/testmqtt/nestjs/sensors/+/temperature')
store(@Payload() reading: Reading) { /* mesaj başına tek replica */ }

Broker'ınızın shared subscription desteklediğini ve @nestjs/microservices sürümünüzün mesajları yönlendirirken öneki kaldırdığını doğrulayın (bir test mesajı gönderin ve yalnızca bir replica'nın logladığını görün). Arka plan: shared subscription'lar.

Servisi uçtan uca test etmek

Her iki handler türünü de hiç client yazmadan deneyebilirsiniz.

1. Uygulamayı MQTT_USER ve MQTT_PASS tanımlı olarak başlatın ve loglarda sorunsuz açıldığını kontrol edin.
2. Online MQTT client'ı açın, broker.connectmq.com adresine WSS ile bağlanın (port 8084, path /mqtt) ve testmqtt/nestjs/# topic'ine abone olun.
3. testmqtt/nestjs/sensors/dev-42/temperature topic'ine {"t": 21.4} publish edin. Nest logunda dev-42: 21.4 °C görünmelidir.
4. testmqtt/nestjs/sum topic'ine Nest zarfında bir istek publish edin:

{"pattern":"testmqtt/nestjs/sum","data":[1,2,3],"id":"manual-test-1"}

1. Milisaniyeler içinde testmqtt/nestjs/sum/reply üzerinde "id":"manual-test-1" ve "response":6 içeren bir mesaj görmelisiniz.

Yanıt hiç gelmiyorsa şu sırayla kontrol edin: pattern string'i iki tarafta da aynı mı (sondaki bir eğik çizgi farklı bir topic demektir), hesabın /reply topic'inde yetkisi var mı, handler exception fırlattı mı (o zaman yanıtta err bulunur). Bağlantı düzeyindeki sorunlar genellikle logda Connection refused: Not authorized veya yeniden bağlanma döngüleri olarak görünür; MQTT.js hataları rehberi bunları anlatır.

NestJS topic'lerini canlı izleyin

Otomatik testler için handler'ları ince tutun ve mantığı doğrudan unit test'lerle test edin; ardından mikroservisi gerçek bir broker'a karşı başlatan küçük bir entegrasyon test paketi ekleyin. Entegrasyon testi rehberi, her test çalıştırmasında topic'leri nasıl izole edeceğinizi gösterir.

Sık karşılaşılan hatalar

NestJS, MQTT'yi herhangi bir transport gibi hissettirir, ama broker semantiği yine geçerlidir: topic'ler, ACL'ler, client ID'ler ve QoS. Bunları doğru kurun, her send() çağrısına timeout ekleyin ve üretimde güvenmeden önce her pattern'i gerçek bir broker ile test edin.

Sıkça sorulan sorular