Tutoriel Angular MQTT avec RxJS, MQTT.js et ngx-mqtt

Angular et MQTT s'accordent naturellement : MQTT pousse les messages, et les observables RxJS sont la manière native d'Angular de traiter des données poussées. Dans ce tutoriel Angular MQTT, vous allez construire un service injectable qui se connecte à un broker en WebSocket sécurisé, expose n'importe quel filtre de topic sous forme d'observable et publie des commandes, puis afficher des valeurs en direct avec le pipe async. Vous verrez aussi comment se positionne la bibliothèque populaire ngx-mqtt, afin de choisir la bonne approche pour votre projet.

Architecture : une connexion, de nombreux observables

L'idée centrale est simple. Un unique service de niveau racine possède le client MQTT.js. Les composants demandent un filtre de topic au service et obtiennent un Observable en retour. Quand le premier composant s'abonne à un filtre, le service envoie un SUBSCRIBE MQTT ; quand le dernier se désabonne, le service envoie un UNSUBSCRIBE.

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

Vous obtenez ainsi exactement un WebSocket par onglet de navigateur, un nettoyage automatique grâce au teardown RxJS et une prise en charge complète des wildcards + et #. Les navigateurs ne peuvent pas utiliser de TCP brut : le service utilise donc wss:// sur le port 8084 avec le chemin /mqtt, comme expliqué dans MQTT over WebSocket.

Mise en place

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

Les exemples utilisent des composants standalone et la fonction inject(), disponibles dans les versions actuelles d'Angular.

Le MqttService

Créez 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";

/** Correspondance des filtres de topic MQTT avec les wildcards + et #. */
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");
      // rétablir tous les abonnements actifs après une (re)connexion
      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 {
        /* conserver le texte */
      }
      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");
  }
}

Comment fonctionne le câblage des observables

• share() combiné au cache de flux permet à plusieurs composants qui observent le même filtre de partager un seul abonnement interne. Deux composants qui observent tous deux plant/+/temp n'entraînent qu'un seul paquet SUBSCRIBE, et le départ de l'un ne coupe pas l'autre.
• finalize s'exécute quand le dernier abonné s'en va : le compteur de références est décrémenté et l'UNSUBSCRIBE MQTT est déclenché, de sorte que le broker cesse d'envoyer des messages que personne n'affiche.
• La correspondance des wildcards se fait côté client avec topicMatches. Le broker filtre déjà selon l'abonnement, mais des filtres qui se recoupent (par exemple plant/# et plant/line1/temp) livrent leurs messages via un seul flux : chaque observable doit donc vérifier quels messages lui appartiennent.
• Détection des changements. Les callbacks de MQTT.js sont déclenchés par des événements WebSocket, que zone.js intercepte : Angular détecte donc les changements automatiquement. Dans les applications zoneless, utilisez le pipe async ou toSignal(), qui marquent tous deux la vue pour mise à jour.

Afficher les données en direct avec le pipe async

Créez un composant dans 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);
  }
}

L'opérateur scan replie le flux en une table du dernier message par topic : le panneau affiche ainsi une carte par appareil au lieu d'une liste qui s'allonge. Connectez-vous ensuite une seule fois au démarrage dans 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");
  }
}

Lancez ng serve, puis publiez des données de test sur le serveur MQTT public gratuit :

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

La carte apparaît immédiatement. L'espace public est lisible par tout le monde : gardez votre propre préfixe aléatoire et n'utilisez que des données de test.

Signals, OnPush et tests

Les versions récentes d'Angular orientent les applications vers les signals et la détection de changements OnPush, et le service à base d'observables s'inscrit dans cette direction sans modification. Convertissez n'importe quel flux à la frontière d'un composant avec toSignal(), par exemple readonly reading = toSignal(this.mqtt.observePayload<Telemetry>(topic), { initialValue: {} }), et lisez reading() dans le template. Le signal se désabonne automatiquement à la destruction du composant, si bien que le comptage de références du service continue de fonctionner.

Pour les tests unitaires, évitez de vous connecter à un vrai broker. Comme les composants ne dépendent que de observe, observePayload, publish et state$, vous pouvez fournir un faux service avec { provide: MqttService, useValue: fake }, où observe renvoie un Subject que vous contrôlez. Poussez des messages dans ce subject et vérifiez le rendu obtenu. Conservez un petit ensemble de tests d'intégration qui se connectent au serveur MQTT public gratuit avec un préfixe aléatoire : ils vérifient de bout en bout le câblage réel, l'URL WebSocket et la structure des topics sans ralentir chaque exécution des tests.

Les retained messages méritent aussi d'être mentionnés ici. Si les appareils publient leur dernière télémétrie avec le flag retain, un composant qui s'abonne après une navigation reçoit immédiatement la valeur actuelle au lieu d'afficher des espaces vides jusqu'au prochain rapport. Le champ retain de MqttMessage vous permet de distinguer cet instantané initial des mises à jour en direct, ce qui est utile pour n'animer que les changements récents ou signaler que la première valeur peut être ancienne.

Enfin, sortez les chaînes de topics des composants. Un petit fichier de constantes ou un constructeur de topics typé fait d'un renommage de topic un changement d'une ligne, et rend triviale la recherche de tous les consommateurs d'un topic dans le code.

Et ngx-mqtt ?

ngx-mqtt est une surcouche Angular de longue date autour de MQTT.js. Elle fournit un MqttService dont la méthode observe(filter) renvoie des observables de IMqttMessage, configuré une seule fois au bootstrap :

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

Comparaison des deux approches :

Si la dernière version de ngx-mqtt prend en charge votre version d'Angular, c'est un démarrage rapide. Si vous voulez les fonctionnalités actuelles de MQTT.js ou moins de contraintes de mise à niveau, le service sur mesure est suffisamment petit pour être entièrement maintenu par vous.

Dépannage des applications Angular MQTT

1. global is not defined dans le navigateur. Une ancienne version de MQTT.js ou une dépendance attend des globales Node. Passez à MQTT.js v5 ou, en solution provisoire, ajoutez (window as any).global = window; en haut de main.ts.
2. Boucles de connexion entre onglets. Un client ID fixe est partagé. Le service en génère un aléatoire, sauf si vous le remplacez.
3. Les messages s'arrêtent après une navigation. Un composant s'est désabonné alors qu'il était le dernier observateur, et le service s'est donc désabonné. C'est voulu ; le prochain observateur se réabonne.
4. Aucune mise à jour dans une application zoneless. Liez les données avec le pipe async ou toSignal() au lieu d'affecter des valeurs dans un callback subscribe manuel.

Configuration de production avec ConnectMQ

En production, créez un broker MQTT managé privé sur ConnectMQ et transmettez les identifiants à connect() : connect("wss://YOUR_BROKER_HOST:8084/mqtt", { username, password, clientId }), avec des valeurs issues du portail ou d'un backend émetteur de jetons. Comme le code frontend est visible par les utilisateurs, utilisez un identifiant dédié au tableau de bord et faites passer les commandes sensibles par votre backend. Un broker MQTT cloud doté d'un certificat Let's Encrypt sur le port WebSocket sécurisé fonctionne dans tous les navigateurs sans configuration supplémentaire.

Tutoriels associés

Les mêmes idées se retrouvent dans le hook React MQTT et le composable Vue MQTT. Pour ajouter des contrats de topics typés par-dessus ce service, lisez MQTT typé avec TypeScript, et pour une vue d'ensemble des options de clients, consultez le guide des bibliothèques clientes MQTT. La documentation RxJS détaille share, scan et finalize.

Démarrer avec ConnectMQ

Connectez dès aujourd'hui votre application Angular à un broker MQTT managé privé avec WebSocket sécurisé. Inscrivez-vous sur app.connectmq.com, choisissez l'offre Serverless gratuite sans carte bancaire, créez un broker et transmettez les identifiants générés au service. Utilisez le client MQTT intégré au navigateur du portail pour publier des messages de test pendant que vous développez vos composants.

Questions fréquentes

Faut-il utiliser ngx-mqtt ou MQTT.js directement dans Angular ?

Utilisez ngx-mqtt si sa version actuelle prend en charge votre version d'Angular et que vous n'avez besoin que des fonctions de base observe et publish. Utilisez MQTT.js directement avec un petit service si vous voulez les fonctionnalités MQTT 5, la dernière version de MQTT.js et un contrôle total sur les mises à niveau.

Angular peut-il se connecter à MQTT en TCP sur le port 1883 ou 8883 ?

Non. Les navigateurs n'autorisent que les connexions WebSocket : les applications Angular utilisent donc wss:// sur le port 8084 avec le chemin /mqtt sur les brokers ConnectMQ.

Comment convertir des messages MQTT en signals Angular ?

Passez l'observable à toSignal() depuis @angular/core/rxjs-interop, par exemple toSignal(mqtt.observePayload<Telemetry>(topic)). Le signal se met à jour à chaque nouveau message.

Chaque composant ouvre-t-il sa propre connexion MQTT ?

Pas avec cette conception. Le service est fourni à la racine : toute l'application partage donc une seule connexion, et chaque filtre de topic observé n'est souscrit qu'une fois, quel que soit le nombre de composants qui l'utilisent.