RxJS, MQTT.js ve ngx-mqtt ile Angular MQTT Eğitimi

Angular ve MQTT doğal olarak birbirine uyar: MQTT mesajları push eder, RxJS observable'ları ise Angular'ın push edilen verileri işlemenin yerel yoludur. Bu Angular MQTT eğitiminde güvenli WebSocket üzerinden bir broker'a bağlanan, herhangi bir topic filtresini observable olarak sunan ve komut yayınlayan, enjekte edilebilir bir servis oluşturacak, ardından canlı değerleri async pipe ile göstereceksiniz. Popüler ngx-mqtt kütüphanesinin nasıl karşılaştırıldığını da görecek, böylece projeniz için doğru yaklaşımı seçebileceksiniz.

Mimari: Tek Bağlantı, Çok Sayıda Observable

Temel fikir basittir. Kök düzeyindeki tek bir servis MQTT.js client'ına sahiptir. Bileşenler servisten bir topic filtresi ister ve karşılığında bir Observable alır. İlk bileşen bir filtreye abone olduğunda servis bir MQTT SUBSCRIBE gönderir; son bileşen aboneliği bıraktığında servis UNSUBSCRIBE gönderir.

Component A --observe('plant/+/temp')--+
Component B --observe('plant/line1/#')-+--> MqttService --wss--> broker.connectmq.com:8084/mqtt
Component C --publish('plant/fan/cmd')-+

Bu size tarayıcı sekmesi başına tam olarak bir WebSocket, RxJS teardown aracılığıyla otomatik temizlik ve + ile # wildcard'ları için tam destek sağlar. Tarayıcılar ham TCP kullanamaz; bu yüzden servis, WebSocket üzerinden MQTT yazısında açıklandığı gibi 8084 portunda /mqtt yoluyla wss:// kullanır.

Kurulum

npm install -g @angular/cli
ng new angular-mqtt-demo --standalone --style=css --routing=false
cd angular-mqtt-demo
npm install mqtt

Örnekler, güncel Angular sürümlerinde bulunan standalone bileşenleri ve inject() fonksiyonunu kullanır.

MqttService

src/app/mqtt.service.ts dosyasını oluşturun:

import { Injectable, OnDestroy } from "@angular/core";
import { BehaviorSubject, Observable, Subject, filter, finalize, map, share } from "rxjs";
import mqtt, { IClientOptions, MqttClient } from "mqtt";

export interface MqttMessage<T = unknown> {
  topic: string;
  payload: T;
  retain: boolean;
  receivedAt: number;
}

export type ConnectionState = "connecting" | "connected" | "reconnecting" | "offline" | "closed";

/** + ve # wildcard'larıyla MQTT topic filtresi eşleştirme. */
export function topicMatches(filterStr: string, topic: string): boolean {
  const f = filterStr.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;
}

@Injectable({ providedIn: "root" })
export class MqttService implements OnDestroy {
  private client?: MqttClient;
  private readonly messages$ = new Subject<MqttMessage>();
  private readonly refCounts = new Map<string, number>();
  private readonly streams = new Map<string, Observable<MqttMessage>>();

  readonly state$ = new BehaviorSubject<ConnectionState>("closed");

  connect(url: string, options: IClientOptions = {}): void {
    if (this.client) return;
    this.state$.next("connecting");
    this.client = mqtt.connect(url, {
      protocolVersion: 5,
      reconnectPeriod: 3000,
      connectTimeout: 10000,
      clientId: `ng-${Math.random().toString(16).slice(2, 10)}`,
      ...options,
    });

    this.client.on("connect", () => {
      this.state$.next("connected");
      // (yeniden) bağlandıktan sonra tüm etkin abonelikleri geri yükle
      const filters = [...this.refCounts.keys()];
      if (filters.length) this.client!.subscribe(filters, { qos: 0 });
    });
    this.client.on("reconnect", () => this.state$.next("reconnecting"));
    this.client.on("offline", () => this.state$.next("offline"));
    this.client.on("error", (err) => console.error("MQTT error", err.message));
    this.client.on("message", (topic, payload, packet) => {
      const text = payload.toString();
      let parsed: unknown = text;
      try {
        parsed = JSON.parse(text);
      } catch {
        /* metin olarak bırak */
      }
      this.messages$.next({ topic, payload: parsed, retain: packet.retain, receivedAt: Date.now() });
    });
  }

  observe<T = unknown>(topicFilter: string): Observable<MqttMessage<T>> {
    let stream = this.streams.get(topicFilter);
    if (!stream) {
      stream = new Observable<MqttMessage>((subscriber) => {
        this.addRef(topicFilter);
        const sub = this.messages$
          .pipe(filter((m) => topicMatches(topicFilter, m.topic)))
          .subscribe(subscriber);
        return () => sub.unsubscribe();
      }).pipe(
        finalize(() => this.removeRef(topicFilter)),
        share()
      );
      this.streams.set(topicFilter, stream);
    }
    return stream as Observable<MqttMessage<T>>;
  }

  observePayload<T = unknown>(topicFilter: string): Observable<T> {
    return this.observe<T>(topicFilter).pipe(map((m) => m.payload));
  }

  publish(topic: string, message: unknown, qos: 0 | 1 | 2 = 0, retain = false): Promise<void> {
    if (!this.client?.connected) return Promise.reject(new Error("MQTT not connected"));
    const body = typeof message === "string" ? message : JSON.stringify(message);
    return this.client.publishAsync(topic, body, { qos, retain }).then(() => undefined);
  }

  private addRef(topicFilter: string): void {
    const count = this.refCounts.get(topicFilter) ?? 0;
    this.refCounts.set(topicFilter, count + 1);
    if (count === 0 && this.client?.connected) this.client.subscribe(topicFilter, { qos: 0 });
  }

  private removeRef(topicFilter: string): void {
    const count = (this.refCounts.get(topicFilter) ?? 1) - 1;
    if (count > 0) {
      this.refCounts.set(topicFilter, count);
      return;
    }
    this.refCounts.delete(topicFilter);
    this.streams.delete(topicFilter);
    if (this.client?.connected) this.client.unsubscribe(topicFilter);
  }

  ngOnDestroy(): void {
    this.client?.end(true);
    this.state$.next("closed");
  }
}

Observable Bağlantıları Nasıl Çalışır?

• share() ve stream önbelleği sayesinde aynı filtreyi izleyen birkaç bileşen tek bir iç aboneliği paylaşır. plant/+/temp filtresini izleyen iki bileşen tek bir SUBSCRIBE paketine neden olur ve birinin ayrılması diğerini kesmez.
• finalize, son abone ayrıldığında çalışır; referans sayısını azaltır ve MQTT UNSUBSCRIBE'ı tetikler, böylece broker kimsenin göstermediği mesajları göndermeyi bırakır.
• Wildcard eşleştirme client tarafında topicMatches ile yapılır. Broker zaten aboneliğe göre filtreleme yapar, ancak örtüşen filtreler (örneğin plant/# ve plant/line1/temp) mesajları tek bir akış üzerinden teslim eder; bu yüzden her observable hangi mesajların kendisine ait olduğunu kontrol etmelidir.
• Değişiklik algılama. MQTT.js callback'leri, zone.js'in yamaladığı WebSocket olaylarından tetiklenir; bu nedenle Angular değişiklikleri otomatik olarak algılar. Zoneless uygulamalarda, her ikisi de görünümü güncelleme için işaretleyen async pipe veya toSignal() kullanın.

Canlı Verileri Async Pipe ile Göstermek

src/app/sensor-panel.component.ts içinde bir bileşen oluşturun:

import { Component, inject } from "@angular/core";
import { AsyncPipe, DatePipe, JsonPipe } from "@angular/common";
import { map, scan } from "rxjs";
import { MqttService, MqttMessage } from "./mqtt.service";

interface Telemetry {
  temperature?: number;
  humidity?: number;
}

const PREFIX = "connectmq/demo/ng-6b19";

@Component({
  selector: "app-sensor-panel",
  standalone: true,
  imports: [AsyncPipe, DatePipe, JsonPipe],
  template: `
    <p>Connection: <strong>{{ mqtt.state$ | async }}</strong></p>
    @if (devices$ | async; as devices) {
      @for (d of devices; track d.topic) {
        <div class="card">
          <h3>{{ d.topic.split('/').at(-2) }}</h3>
          <p>{{ d.payload.temperature ?? '--' }} &deg;C, {{ d.payload.humidity ?? '--' }} %</p>
          <small>{{ d.receivedAt | date: 'mediumTime' }}</small>
        </div>
      }
    }
    <button (click)="toggle()" [disabled]="(mqtt.state$ | async) !== 'connected'">Toggle pump1</button>
  `,
})
export class SensorPanelComponent {
  readonly mqtt = inject(MqttService);

  readonly devices$ = this.mqtt.observe<Telemetry>(`${PREFIX}/+/telemetry`).pipe(
    scan((acc, msg) => {
      const next = new Map(acc);
      next.set(msg.topic, msg);
      return next;
    }, new Map<string, MqttMessage<Telemetry>>()),
    map((byTopic) => [...byTopic.values()])
  );

  toggle(): void {
    this.mqtt.publish(`${PREFIX}/pump1/cmd`, { action: "toggle" }, 1).catch(console.error);
  }
}

scan operatörü akışı topic başına son mesajı içeren bir haritaya indirger; böylece panel büyüyen bir liste yerine cihaz başına bir kart gösterir. Ardından src/app/app.component.ts içinde başlangıçta bir kez bağlanın:

import { Component, inject } from "@angular/core";
import { MqttService } from "./mqtt.service";
import { SensorPanelComponent } from "./sensor-panel.component";

@Component({
  selector: "app-root",
  standalone: true,
  imports: [SensorPanelComponent],
  template: `<h1>Angular MQTT Demo</h1><app-sensor-panel />`,
})
export class AppComponent {
  constructor() {
    inject(MqttService).connect("wss://broker.connectmq.com:8084/mqtt");
  }
}

ng serve çalıştırın, ardından public MQTT test broker'a test verisi yayınlayın:

mosquitto_pub -h broker.connectmq.com -p 1883 -i ng-pub-6b19 \
  -t connectmq/demo/ng-6b19/pump1/telemetry -m '{"temperature": 36.5, "humidity": 48}'

Kart hemen görünür. Public alan herkes tarafından okunabilir; bu yüzden kendi rastgele önekinizi ve yalnızca test verisi kullanın.

Signal'ler, OnPush ve Test

Son Angular sürümleri uygulamaları signal'lere ve OnPush değişiklik algılamaya yönlendirir; observable tabanlı servis de bu yöne hiçbir değişiklik gerektirmeden uyar. Herhangi bir akışı bileşenin sınırında toSignal() ile dönüştürün; örneğin readonly reading = toSignal(this.mqtt.observePayload<Telemetry>(topic), { initialValue: {} }) yazıp şablonda reading() okuyun. Signal, bileşen yok edildiğinde aboneliği otomatik olarak bırakır; dolayısıyla servisteki referans sayımı çalışmaya devam eder.

Birim testlerinde gerçek bir broker'a bağlanmaktan kaçının. Bileşenler yalnızca observe, observePayload, publish ve state$ öğelerine bağımlı olduğundan, observe metodunun sizin kontrol ettiğiniz bir Subject döndürdüğü sahte bir servisi { provide: MqttService, useValue: fake } ile sağlayabilirsiniz. Subject'e mesajlar gönderin ve render edilen çıktıyı doğrulayın. Rastgele bir önekle public broker'a bağlanan küçük bir entegrasyon testi seti tutun; bu, her test çalıştırmasını yavaşlatmadan gerçek bağlantıyı, WebSocket URL'sini ve topic yapısını uçtan uca doğrular.

Retained message'lar da burada anılmayı hak ediyor. Cihazlar son telemetrilerini retain bayrağıyla yayınlıyorsa, sayfa geçişinden sonra abone olan bir bileşen bir sonraki rapora kadar boş yer tutucular göstermek yerine güncel değeri hemen alır. MqttMessage üzerindeki retain alanı bu ilk anlık görüntüyü canlı güncellemelerden ayırt etmenizi sağlar; bu da yalnızca yeni değişiklikleri canlandırmak veya ilk değeri muhtemelen eski olarak işaretlemek istediğinizde kullanışlıdır.

Son olarak topic dizelerini bileşenlerin dışında tutun. Küçük bir sabitler dosyası veya tipli bir topic oluşturucu, yeniden adlandırılan bir topic'in tek satırlık bir değişiklik olması anlamına gelir ve bir topic'in tüm tüketicilerini kod tabanında aramayı çok kolaylaştırır.

Peki ngx-mqtt?

ngx-mqtt, MQTT.js etrafında uzun süredir var olan bir Angular sarmalayıcısıdır. Başlangıçta bir kez yapılandırılan ve observe(filter) ile IMqttMessage observable'ları döndüren bir MqttService sunar:

import { importProvidersFrom } from "@angular/core";
import { bootstrapApplication } from "@angular/platform-browser";
import { IMqttServiceOptions, MqttModule } from "ngx-mqtt";
import { AppComponent } from "./app/app.component";

const MQTT_OPTIONS: IMqttServiceOptions = {
  hostname: "broker.connectmq.com",
  port: 8084,
  path: "/mqtt",
  protocol: "wss",
  clientId: `ngx-${Math.random().toString(16).slice(2, 10)}`,
};

bootstrapApplication(AppComponent, {
  providers: [importProvidersFrom(MqttModule.forRoot(MQTT_OPTIONS))],
});

İki yaklaşımın karşılaştırması:

ngx-mqtt'nin son sürümü Angular sürümünüzü destekliyorsa hızlı bir başlangıçtır. Güncel MQTT.js özelliklerini veya daha az yükseltme kısıtlamasını istiyorsanız özel servis tamamen sahiplenebileceğiniz kadar küçüktür.

Angular MQTT Uygulamalarında Sorun Giderme

1. Tarayıcıda global is not defined. Eski bir MQTT.js veya bir bağımlılık Node global'lerini bekliyor. MQTT.js v5'e yükseltin veya geçici çözüm olarak main.ts dosyasının en üstüne (window as any).global = window; ekleyin.
2. Sekmeler arasında bağlantı döngüleri. Sabit bir client ID paylaşılıyor. Servis, siz geçersiz kılmadığınız sürece rastgele bir ID üretir.
3. Sayfa geçişinden sonra mesajlar duruyor. Bir bileşen aboneliği bıraktı ve son gözlemciydi, bu yüzden servis de aboneliği bıraktı. Bu tasarım gereğidir; bir sonraki gözlemci yeniden abone olur.
4. Zoneless bir uygulamada güncelleme yok. Değerleri manuel bir subscribe callback'inde atamak yerine async pipe veya toSignal() ile bağlayın.

ConnectMQ ile Üretim Kurulumu

Üretim için ConnectMQ üzerinde private bir broker oluşturun ve kimlik bilgilerini connect() metoduna iletin: portaldan veya token üreten bir backend'den gelen değerlerle connect("wss://YOUR_BROKER_HOST:8084/mqtt", { username, password, clientId }). Frontend kodu kullanıcılar tarafından görülebildiği için özel bir pano kimlik bilgisi kullanın ve hassas komutları backend'iniz üzerinden yönlendirin. Güvenli WebSocket portunda Let's Encrypt sertifikası olan bir bulut MQTT broker, ek kurulum gerektirmeden her tarayıcıda çalışır.

İlgili Eğitimler

Aynı fikirler React MQTT hook'u ve Vue MQTT composable'ı yazılarında da yer alır. Bu servisin üzerine tipli topic sözleşmeleri eklemek için TypeScript ile tip güvenli MQTT yazısını okuyun; client seçeneklerine daha geniş bir bakış için MQTT client kütüphaneleri rehberine göz atın. RxJS dokümantasyonu share, scan ve finalize operatörlerini ayrıntılı olarak ele alır.

ConnectMQ ile Başlayın

Angular uygulamanızı bugün güvenli WebSocket destekli private bir broker'a bağlayın. app.connectmq.com adresinden kaydolun, kredi kartı gerektirmeyen ücretsiz Serverless planını seçin, bir broker oluşturun ve üretilen kimlik bilgilerini servise iletin. Bileşenlerinizi geliştirirken test mesajları yayınlamak için portalın tarayıcı içi MQTT client'ını kullanın.

Sık Sorulan Sorular

Angular'da ngx-mqtt mi yoksa doğrudan MQTT.js mi kullanmalıyım?

Güncel sürümü Angular sürümünüzü destekliyorsa ve yalnızca temel observe ve publish işlemlerine ihtiyacınız varsa ngx-mqtt kullanın. MQTT 5 özelliklerini, en yeni MQTT.js'i ve yükseltmeler üzerinde tam kontrolü istiyorsanız küçük bir servisle doğrudan MQTT.js kullanın.

Angular, 1883 veya 8883 TCP portu üzerinden MQTT'ye bağlanabilir mi?

Hayır. Tarayıcılar yalnızca WebSocket bağlantılarına izin verir; bu yüzden Angular uygulamaları ConnectMQ broker'larında 8084 portunda /mqtt yoluyla wss:// kullanır.

MQTT mesajlarını Angular signal'lerine nasıl dönüştürürüm?

Observable'ı @angular/core/rxjs-interop içindeki toSignal() fonksiyonuna verin; örneğin toSignal(mqtt.observePayload<Telemetry>(topic)). Signal, her yeni mesaj geldiğinde güncellenir.

Her bileşen kendi MQTT bağlantısını mı açar?

Bu tasarımda hayır. Servis root düzeyinde sağlanır; böylece tüm uygulama tek bir bağlantıyı paylaşır ve gözlemlenen her topic filtresine, kaç bileşenin kullandığından bağımsız olarak bir kez abone olunur.