MQTT in Angular mit ngx-mqtt und MQTT.js

npm i ngx-mqtt

import { ApplicationConfig, importProvidersFrom } from '@angular/core';
import { IMqttServiceOptions, MqttModule } from 'ngx-mqtt';

const MQTT_OPTIONS: IMqttServiceOptions = {
  hostname: 'broker.connectmq.com',
  port: 8084,
  protocol: 'wss',
  path: '/mqtt',
  username: 'username',
  password: 'password',
};

export const appConfig: ApplicationConfig = {
  providers: [importProvidersFrom(MqttModule.forRoot(MQTT_OPTIONS))],
};

import { Component, inject } from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { IMqttMessage, MqttService } from 'ngx-mqtt';
import { map } from 'rxjs';

@Component({
  selector: 'app-status',
  template: `<p>Last status: {{ status() }}</p>
             <button (click)="ping()">Ping</button>`,
})
export class StatusComponent {
  private readonly mqtt = inject(MqttService);

  readonly status = toSignal(
    this.mqtt.observe('testmqtt/angular/status').pipe(map((m: IMqttMessage) => m.payload.toString())),
    { initialValue: '—' },
  );

  ping() {
    this.mqtt.unsafePublish('testmqtt/angular/cmd/ping', String(Date.now()), { qos: 1 });
  }
}

npm i mqtt

import { Injectable, OnDestroy, PLATFORM_ID, inject } from '@angular/core';
import { isPlatformBrowser } from '@angular/common';
import mqtt, { type MqttClient } from 'mqtt';
import { BehaviorSubject, Observable, Subject, filter, map } from 'rxjs';
import { environment } from '../environments/environment';

export type MqttState = 'idle' | 'connecting' | 'connected' | 'reconnecting' | 'offline' | 'error';
export interface MqttMessage<T> { topic: string; data: T }

export function matches(topicFilter: string, topic: string): boolean {
  const f = topicFilter.split('/');
  const t = topic.split('/');
  for (let i = 0; i < f.length; i++) {
    if (f[i] === '#') return true;
    if (i >= t.length || (f[i] !== '+' && f[i] !== t[i])) return false;
  }
  return f.length === t.length;
}

@Injectable({ providedIn: 'root' })
export class MqttClientService implements OnDestroy {
  readonly state$ = new BehaviorSubject<MqttState>('idle');
  private readonly incoming$ = new Subject<{ topic: string; payload: Uint8Array }>();
  private readonly refCounts = new Map<string, number>();
  private client: MqttClient | null = null;

  constructor() {
    if (!isPlatformBrowser(inject(PLATFORM_ID))) return; // kein MQTT beim SSR
    this.state$.next('connecting');
    this.client = mqtt.connect(environment.mqtt.url, {
      username: environment.mqtt.username,
      password: environment.mqtt.password,
      clientId: `ng-${crypto.randomUUID().slice(0, 8)}`,
      reconnectPeriod: 2000,
      connectTimeout: 10_000,
    });
    this.client.on('connect', () => this.state$.next('connected'));
    this.client.on('reconnect', () => this.state$.next('reconnecting'));
    this.client.on('offline', () => this.state$.next('offline'));
    this.client.on('error', () => this.state$.next('error'));
    this.client.on('message', (topic, payload) => this.incoming$.next({ topic, payload }));
  }

  observe<T = unknown>(topicFilter: string): Observable<MqttMessage<T>> {
    return new Observable<MqttMessage<T>>((subscriber) => {
      this.retain(topicFilter);
      const sub = this.incoming$
        .pipe(
          filter((m) => matches(topicFilter, m.topic)),
          map((m) => {
            const text = new TextDecoder().decode(m.payload);
            let data: T;
            try { data = JSON.parse(text) as T; } catch { data = text as T; }
            return { topic: m.topic, data };
          }),
        )
        .subscribe(subscriber);
      return () => {
        sub.unsubscribe();
        this.release(topicFilter);
      };
    });
  }

  async publish(topic: string, message: string, retain = false): Promise<void> {
    await this.client?.publishAsync(topic, message, { qos: 1, retain });
  }

  ngOnDestroy(): void {
    this.client?.end(true);
  }

  private retain(topicFilter: string) {
    const n = this.refCounts.get(topicFilter) ?? 0;
    if (n === 0) this.client?.subscribe(topicFilter, { qos: 1 });
    this.refCounts.set(topicFilter, n + 1);
  }

  private release(topicFilter: string) {
    const n = (this.refCounts.get(topicFilter) ?? 1) - 1;
    if (n === 0) {
      this.refCounts.delete(topicFilter);
      this.client?.unsubscribe(topicFilter);
    } else {
      this.refCounts.set(topicFilter, n);
    }
  }
}

import { Component, inject } from '@angular/core';
import { DecimalPipe } from '@angular/common';
import { toSignal } from '@angular/core/rxjs-interop';
import { auditTime, map, scan } from 'rxjs';
import { MqttClientService } from './mqtt-client.service';

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

@Component({
  selector: 'app-sensors',
  imports: [DecimalPipe],
  template: `
    <p>Broker: {{ state() }}</p>
    <table>
      @for (row of rows(); track row.device) {
        <tr><td>{{ row.device }}</td><td>{{ row.t | number: '1.1-1' }} °C</td></tr>
      }
    </table>
    <button [disabled]="state() !== 'connected'" (click)="ping()">Ping</button>
  `,
})
export class SensorsComponent {
  private readonly mqtt = inject(MqttClientService);

  readonly state = toSignal(this.mqtt.state$, { requireSync: true });

  readonly rows = toSignal(
    this.mqtt.observe<Reading>('testmqtt/angular/sensors/+').pipe(
      scan((acc, m) => ({ ...acc, [m.topic.split('/').pop()!]: m.data.t }), {} as Record<string, number>),
      auditTime(100),
      map((latest) => Object.entries(latest).map(([device, t]) => ({ device, t }))),
    ),
    { initialValue: [] },
  );

  ping() {
    this.mqtt.publish('testmqtt/angular/cmd/ping', String(Date.now()));
  }
}

import { TestBed } from '@angular/core/testing';
import { BehaviorSubject, Subject } from 'rxjs';
import { MqttClientService, MqttMessage } from './mqtt-client.service';
import { SensorsComponent } from './sensors.component';

it('renders the latest reading per device', async () => {
  const messages = new Subject<MqttMessage<{ t: number }>>();
  await TestBed.configureTestingModule({
    imports: [SensorsComponent],
    providers: [{
      provide: MqttClientService,
      useValue: { state$: new BehaviorSubject('connected'), observe: () => messages, publish: async () => {} },
    }],
  }).compileComponents();

  const fixture = TestBed.createComponent(SensorsComponent);
  fixture.detectChanges();
  messages.next({ topic: 'testmqtt/angular/sensors/room-1', data: { t: 21 } });
  await new Promise((r) => setTimeout(r, 150)); // auf auditTime warten
  fixture.detectChanges();
  expect(fixture.nativeElement.textContent).toContain('room-1');
});