Angular MQTT Tutorial mit RxJS und ngx-mqtt

Q: Wie wandle ich MQTT-Nachrichten in Angular Signals um?

Übergeben Sie das Observable an toSignal() aus @angular/core/rxjs-interop, zum Beispiel toSignal(mqtt.observePayload (topic)). Das Signal aktualisiert sich bei jeder neuen Nachricht.

Angular und MQTT passen von Natur aus zusammen: MQTT liefert Nachrichten per Push, und RxJS-Observables sind Angulars native Art, gepushte Daten zu verarbeiten. In diesem Angular-MQTT-Tutorial bauen Sie einen injizierbaren Service, der sich über sicheren WebSocket mit einem Broker verbindet, jeden Topic-Filter als Observable bereitstellt und Befehle publiziert. Anschließend rendern Sie Live-Werte mit der async-Pipe. Außerdem sehen Sie, wie die beliebte Bibliothek ngx-mqtt im Vergleich abschneidet, damit Sie den richtigen Ansatz für Ihr Projekt wählen können.

Architektur: eine Verbindung, viele Observables

Die Grundidee ist einfach. Ein einziger Service auf Root-Ebene besitzt den MQTT.js-Client. Komponenten fragen beim Service einen Topic-Filter an und erhalten ein Observable zurück. Abonniert die erste Komponente einen Filter, sendet der Service ein MQTT-SUBSCRIBE; meldet sich die letzte ab, sendet er UNSUBSCRIBE.

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

So erhalten Sie genau einen WebSocket pro Browser-Tab, automatisches Aufräumen über das Teardown von RxJS und volle Unterstützung für die Wildcards + und #. Browser können kein rohes TCP nutzen, daher verwendet der Service wss:// auf Port 8084 mit dem Pfad /mqtt, wie in MQTT over WebSocket erklärt.

Setup

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

Die Beispiele verwenden Standalone-Komponenten und die Funktion inject(), die in aktuellen Angular-Versionen verfügbar sind.

Der MqttService

Erstellen Sie src/app/mqtt.service.ts:

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

/** MQTT-Topic-Filter-Matching mit den Wildcards + und #. */
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");
      // alle aktiven Subscriptions nach (Re-)Connect wiederherstellen
      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 {
        /* Text beibehalten */
      }
      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");
  }
}

Wie die Observable-Verdrahtung funktioniert

share() plus Stream-Cache bedeutet, dass sich mehrere Komponenten, die denselben Filter beobachten, eine innere Subscription teilen. Zwei Komponenten, die beide plant/+/temp beobachten, erzeugen ein einziges SUBSCRIBE-Paket, und wenn eine davon wegfällt, wird die andere nicht abgeschnitten.
finalize läuft, wenn der letzte Subscriber geht. Es verringert den Referenzzähler und löst das MQTT-UNSUBSCRIBE aus, sodass der Broker keine Nachrichten mehr sendet, die niemand anzeigt.
Wildcard-Matching erfolgt clientseitig mit topicMatches. Der Broker filtert zwar bereits nach Subscription, aber überlappende Filter (etwa plant/# und plant/line1/temp) liefern Nachrichten über einen Stream, daher muss jedes Observable prüfen, welche Nachrichten zu ihm gehören.
Change Detection. MQTT.js-Callbacks werden durch WebSocket-Events ausgelöst, die zone.js patcht, sodass Angular Änderungen automatisch erkennt. In zonenlosen Anwendungen verwenden Sie die async-Pipe oder toSignal(), die beide die View zur Aktualisierung markieren.

Live-Daten mit der async-Pipe rendern

Erstellen Sie eine Komponente in src/app/sensor-panel.component.ts:

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

Der Operator scan faltet den Stream zu einer Map mit der letzten Nachricht pro Topic, sodass das Panel eine Karte pro Gerät anzeigt statt einer wachsenden Liste. Verbinden Sie sich dann einmalig beim Start in src/app/app.component.ts:

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

Starten Sie ng serve und publizieren Sie Testdaten an den öffentlichen MQTT-Broker:

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

Die Karte erscheint sofort. Der öffentliche MQTT-Broker ist für alle lesbar und nur für Tests und Prototypen gedacht, behalten Sie also Ihr eigenes zufälliges Präfix bei und verwenden Sie keine Produktionsdaten.

Signals, OnPush und Tests

Aktuelle Angular-Versionen lenken Anwendungen in Richtung Signals und OnPush-Change-Detection, und der Observable-Service passt ohne Änderungen dazu. Wandeln Sie einen beliebigen Stream am Rand einer Komponente mit toSignal() um, zum Beispiel readonly reading = toSignal(this.mqtt.observePayload<Telemetry>(topic), { initialValue: {} }), und lesen Sie reading() im Template. Das Signal meldet sich automatisch ab, wenn die Komponente zerstört wird, sodass die Referenzzählung im Service weiterhin funktioniert.

Verbinden Sie sich in Unit-Tests nicht mit einem echten Broker. Da Komponenten nur von observe, observePayload, publish und state$ abhängen, können Sie mit { provide: MqttService, useValue: fake } einen Fake-Service bereitstellen, bei dem observe ein Subject zurückgibt, das Sie kontrollieren. Schieben Sie Nachrichten in das Subject und prüfen Sie die gerenderte Ausgabe. Behalten Sie einige wenige Integrationstests, die sich mit zufälligem Präfix mit dem öffentlichen MQTT-Broker verbinden – sie verifizieren die echte Verdrahtung, die WebSocket-URL und die Topic-Struktur durchgängig, ohne jeden Testlauf auszubremsen.

Auch Retained Messages verdienen hier Erwähnung. Publizieren Geräte ihre letzte Telemetrie mit Retain-Flag, erhält eine Komponente, die nach einer Navigation abonniert, sofort den aktuellen Wert, statt bis zur nächsten Meldung leere Platzhalter zu zeigen. Mit dem Feld retain in MqttMessage können Sie diesen anfänglichen Snapshot von Live-Updates unterscheiden – nützlich, wenn Sie nur frische Änderungen animieren oder den ersten Wert als möglicherweise veraltet kennzeichnen möchten.

Halten Sie schließlich Topic-Strings aus den Komponenten heraus. Eine kleine Konstantendatei oder ein typisierter Topic-Builder macht ein umbenanntes Topic zu einer einzeiligen Änderung und die Suche nach allen Konsumenten eines Topics in der Codebasis trivial. Für ein Leitstand- oder OEE-Dashboard in der Fertigung bildet ein solcher Builder idealerweise die ISA-95-Hierarchie eines Unified Namespace (UNS) ab, etwa acme/werk-stuttgart/montage/linie-1/zelle-3/telemetry.

Und was ist mit ngx-mqtt?

ngx-mqtt ist ein langjähriger Angular-Wrapper um MQTT.js. Er stellt einen MqttService bereit, dessen observe(filter) Observables von IMqttMessage zurückgibt, und wird einmalig beim Bootstrap konfiguriert:

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

So schneiden die beiden Ansätze im Vergleich ab:

Aspekt	Eigener Service (dieser Artikel)	ngx-mqtt
Abhängigkeiten	Nur MQTT.js	ngx-mqtt plus dessen fixierte MQTT.js-Version
MQTT-5-Features	Alles, was Ihre MQTT.js-Version unterstützt	Abhängig von der mitgelieferten MQTT.js-Version
Angular-Versions-Updates	Sie steuern die Kompatibilität	Sie müssen auf Releases der Bibliothek warten
Eigener Code	Rund 120 Zeilen	Nur Konfiguration
Wildcard-Observables und Referenzzählung	Ja	Ja

Unterstützt das neueste Release von ngx-mqtt Ihre Angular-Version, ist es ein schneller Einstieg. Wollen Sie aktuelle MQTT.js-Features oder weniger Einschränkungen bei Upgrades, ist der eigene Service klein genug, um ihn vollständig selbst zu pflegen.

Fehlerbehebung bei Angular-MQTT-Apps

global is not defined im Browser. Eine ältere MQTT.js-Version oder eine Abhängigkeit erwartet Node-Globals. Aktualisieren Sie auf MQTT.js v5 oder fügen Sie als Notlösung (window as any).global = window; am Anfang von main.ts ein.
Verbindungsschleifen zwischen Tabs. Eine feste Client ID wird geteilt. Der Service erzeugt eine zufällige, sofern Sie sie nicht überschreiben.
Nachrichten bleiben nach einer Navigation aus. Eine Komponente hat sich abgemeldet und war der letzte Beobachter, also hat der Service das Topic abbestellt. Das ist beabsichtigt; der nächste Beobachter abonniert erneut.
Keine Updates in einer zonenlosen App. Binden Sie mit der async-Pipe oder toSignal(), statt Werte in einem manuellen subscribe-Callback zuzuweisen.

Produktions-Setup mit ConnectMQ

Für die Produktion erstellen Sie einen privaten Broker auf ConnectMQ und übergeben Zugangsdaten an connect(): connect("wss://YOUR_BROKER_HOST:8084/mqtt", { username, password, clientId }), mit Werten aus dem Portal oder einem Backend, das Tokens ausstellt. Da Frontend-Code für Nutzer sichtbar ist, verwenden Sie eigene Zugangsdaten für das Dashboard und leiten sensible Befehle über Ihr Backend. Ein Cloud-MQTT-Broker mit Let's-Encrypt-Zertifikat auf dem sicheren WebSocket-Port funktioniert in jedem Browser ohne zusätzliche Einrichtung.

Starten Sie mit ConnectMQ

Verbinden Sie Ihre Angular-App noch heute mit einem privaten Broker mit sicherem WebSocket. Registrieren Sie sich unter app.connectmq.com, wählen Sie den kostenlosen Serverless-Plan ohne Kreditkarte, erstellen Sie einen Broker und übergeben Sie die generierten Zugangsdaten an den Service. Nutzen Sie den MQTT-Client im Browser des Portals, um Testnachrichten zu publizieren, während Sie Ihre Komponenten bauen.

Häufig gestellte Fragen

Sollte ich in Angular ngx-mqtt oder MQTT.js direkt verwenden?

Verwenden Sie ngx-mqtt, wenn dessen aktuelles Release Ihre Angular-Version unterstützt und Sie nur einfaches Observe und Publish brauchen. Verwenden Sie MQTT.js direkt mit einem kleinen Service, wenn Sie MQTT-5-Features, das neueste MQTT.js und volle Kontrolle über Upgrades möchten.

Kann sich Angular über TCP-Port 1883 oder 8883 mit MQTT verbinden?

Nein. Browser erlauben nur WebSocket-Verbindungen, daher nutzen Angular-Apps auf ConnectMQ-Brokern wss:// auf Port 8084 mit dem Pfad /mqtt.

Wie wandle ich MQTT-Nachrichten in Angular Signals um?

Übergeben Sie das Observable an toSignal() aus @angular/core/rxjs-interop, zum Beispiel toSignal(mqtt.observePayload<Telemetry>(topic)). Das Signal aktualisiert sich bei jeder neuen Nachricht.

Öffnet jede Komponente ihre eigene MQTT-Verbindung?

Nicht mit diesem Design. Der Service wird in Root bereitgestellt, sodass die gesamte Anwendung eine Verbindung teilt und jeder beobachtete Topic-Filter genau einmal abonniert wird – unabhängig davon, wie viele Komponenten ihn nutzen.

Angular MQTT Tutorial mit RxJS, MQTT.js und ngx-mqtt