Tutorial de Angular y MQTT con RxJS, MQTT.js y ngx-mqtt

Angular y MQTT encajan de forma natural: MQTT envía mensajes push y los observables de RxJS son la forma nativa de Angular de gestionar datos recibidos por push. En este tutorial de Angular y MQTT crearás un servicio inyectable que se conecta a un broker mediante WebSocket seguro, expone cualquier filtro de topic como un observable y publica comandos, y después mostrarás valores en vivo con el pipe async. También verás cómo se compara con la popular librería ngx-mqtt, para que elijas el enfoque adecuado para tu proyecto.

Arquitectura: una conexión, muchos observables

La idea central es sencilla. Un único servicio a nivel raíz es dueño del cliente MQTT.js. Los componentes piden al servicio un filtro de topic y reciben un Observable. Cuando el primer componente se suscribe a un filtro, el servicio envía un SUBSCRIBE de MQTT; cuando el último se da de baja, el servicio envía 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')-+

Así obtienes exactamente un WebSocket por pestaña del navegador, limpieza automática mediante el teardown de RxJS y soporte completo de los wildcards + y #. Los navegadores no pueden usar TCP directo, así que el servicio usa wss:// en el puerto 8084 con la ruta /mqtt, como se explica en MQTT sobre WebSocket.

Configuración

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

Los ejemplos usan componentes standalone y la función inject(), disponibles en las versiones actuales de Angular.

El MqttService

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

/** Coincidencia de filtros de topic MQTT con los wildcards + y #. */
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");
      // restaura todas las suscripciones activas tras (re)conectar
      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 {
        /* conservar el texto */
      }
      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");
  }
}

Cómo funciona el cableado de observables

• share() junto con la caché de streams hace que varios componentes que observan el mismo filtro compartan una única suscripción interna. Dos componentes que observan plant/+/temp generan un solo paquete SUBSCRIBE, y que uno se vaya no corta al otro.
• finalize se ejecuta cuando se va el último suscriptor, lo que decrementa el contador de referencias y dispara el UNSUBSCRIBE de MQTT, de modo que el broker deja de enviar mensajes que nadie muestra.
• La coincidencia de wildcards se hace en el cliente con topicMatches. El broker ya filtra por suscripción, pero los filtros que se solapan (por ejemplo plant/# y plant/line1/temp) entregan mensajes a través de un único stream, así que cada observable debe comprobar qué mensajes le pertenecen.
• Detección de cambios. Los callbacks de MQTT.js se disparan desde eventos de WebSocket, que zone.js parchea, así que Angular detecta los cambios automáticamente. En aplicaciones sin zone, usa el pipe async o toSignal(), que marcan la vista para actualizarse.

Mostrar datos en vivo con el pipe async

Crea un componente en 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);
  }
}

El operador scan acumula el stream en un mapa con el último mensaje de cada topic, así que el panel muestra una tarjeta por dispositivo en lugar de una lista que crece. Después, conéctate una sola vez al arrancar en 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");
  }
}

Ejecuta ng serve y publica datos de prueba en el servidor MQTT público gratuito de ConnectMQ, pensado para pruebas y prototipos, no para datos de producció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}'

La tarjeta aparece de inmediato. El área pública la puede leer cualquiera, así que mantén tu propio prefijo aleatorio y usa solo datos de prueba.

Signals, OnPush y pruebas

Las versiones recientes de Angular empujan las aplicaciones hacia los signals y la detección de cambios OnPush, y el servicio basado en observables encaja en esa dirección sin cambios. Convierte cualquier stream en el borde de un componente con toSignal(), por ejemplo readonly reading = toSignal(this.mqtt.observePayload<Telemetry>(topic), { initialValue: {} }), y lee reading() en la plantilla. El signal se da de baja automáticamente cuando se destruye el componente, así que el conteo de referencias del servicio sigue funcionando.

En las pruebas unitarias, evita conectarte a un broker real. Como los componentes solo dependen de observe, observePayload, publish y state$, puedes proporcionar un servicio falso con { provide: MqttService, useValue: fake } en el que observe devuelva un Subject que controlas. Empuja mensajes al subject y comprueba la salida renderizada. Mantén un pequeño conjunto de pruebas de integración que se conecten al servidor público con un prefijo aleatorio, lo que verifica de extremo a extremo el cableado real, la URL del WebSocket y la estructura de topics sin ralentizar cada ejecución de pruebas.

Los retained messages también merecen una mención. Si los dispositivos publican su última telemetría con el flag retain, un componente que se suscribe tras una navegación recibe el valor actual al momento en lugar de mostrar marcadores vacíos hasta el siguiente informe. El campo retain de MqttMessage te permite distinguir esa instantánea inicial de las actualizaciones en vivo, algo útil cuando quieres animar solo los cambios nuevos o marcar el primer valor como posiblemente antiguo.

Por último, saca las cadenas de topic de los componentes. Un pequeño archivo de constantes o un constructor de topics tipado hace que renombrar un topic sea un cambio de una línea, y convierte en trivial buscar en el código todos los consumidores de un topic.

¿Y ngx-mqtt?

ngx-mqtt es un envoltorio veterano de Angular sobre MQTT.js. Proporciona un MqttService con observe(filter) que devuelve observables de IMqttMessage, configurado una vez en el arranque:

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

Cómo se comparan los dos enfoques:

Si la última versión de ngx-mqtt admite tu versión de Angular, es un arranque rápido. Si quieres las funciones actuales de MQTT.js o menos restricciones al actualizar, el servicio propio es lo bastante pequeño como para mantenerlo tú mismo.

Resolución de problemas en apps Angular con MQTT

1. global is not defined en el navegador. Una versión antigua de MQTT.js o una dependencia espera los globales de Node. Actualiza a MQTT.js v5 o, como solución temporal, añade (window as any).global = window; al principio de main.ts.
2. Bucles de conexión entre pestañas. Se comparte un client ID fijo. El servicio genera uno aleatorio salvo que lo sobrescribas.
3. Los mensajes se detienen tras navegar. Un componente se dio de baja y era el último observador, así que el servicio canceló la suscripción. Es intencionado; el siguiente observador vuelve a suscribirse.
4. No hay actualizaciones en una app sin zone. Enlaza con el pipe async o toSignal() en lugar de asignar valores en un callback subscribe manual.

Configuración para producción con ConnectMQ

En producción, crea un broker privado en un broker MQTT gestionado como ConnectMQ y pasa las credenciales a connect(): connect("wss://YOUR_BROKER_HOST:8084/mqtt", { username, password, clientId }), usando valores del portal o de un backend que emita tokens. Como el código del frontend es visible para los usuarios, usa una credencial dedicada para el dashboard y envía los comandos sensibles a través de tu backend. Un broker MQTT en la nube con un certificado de Let's Encrypt en el puerto de WebSocket seguro funciona en cualquier navegador sin configuración adicional.

Tutoriales relacionados

Las mismas ideas aparecen en el hook de React para MQTT y en el composable de Vue para MQTT. Para añadir contratos de topics tipados sobre este servicio, lee MQTT con tipos seguros en TypeScript, y para una visión más amplia de las opciones de cliente, consulta la guía de librerías cliente MQTT. La documentación de RxJS explica share, scan y finalize en detalle.

Empieza con ConnectMQ

Conecta hoy tu app Angular a un broker privado con WebSocket seguro. Regístrate en app.connectmq.com, elige el plan Serverless gratuito sin tarjeta de crédito, crea un broker y pasa las credenciales generadas al servicio. Usa el cliente MQTT del portal en el navegador para publicar mensajes de prueba mientras construyes tus componentes.

Preguntas frecuentes

¿Debo usar ngx-mqtt o MQTT.js directamente en Angular?

Usa ngx-mqtt si su versión actual admite tu versión de Angular y solo necesitas observar y publicar de forma básica. Usa MQTT.js directamente con un pequeño servicio cuando quieras las funciones de MQTT 5, la versión más reciente de MQTT.js y control total sobre las actualizaciones.

¿Puede Angular conectarse a MQTT por TCP en el puerto 1883 u 8883?

No. Los navegadores solo permiten conexiones WebSocket, así que las apps Angular usan wss:// en el puerto 8084 con la ruta /mqtt en los brokers de ConnectMQ.

¿Cómo convierto mensajes MQTT en signals de Angular?

Pasa el observable a toSignal() de @angular/core/rxjs-interop, por ejemplo toSignal(mqtt.observePayload<Telemetry>(topic)). El signal se actualiza cada vez que llega un mensaje nuevo.

¿Cada componente abre su propia conexión MQTT?

No con este diseño. El servicio se proporciona en root, así que toda la aplicación comparte una conexión, y cada filtro de topic observado se suscribe una sola vez sin importar cuántos componentes lo usen.