MQTT en Angular: ngx-mqtt y un servicio MQTT.js con RxJS

Hay dos buenas formas de usar MQTT en Angular. ngx-mqtt envuelve MQTT.js en un módulo Angular y te da MqttService.observe(), que devuelve observables RxJS: la vía más rápida para llevar mensajes a un componente. O escribes tú mismo un pequeño servicio inyectable sobre MQTT.js, unas sesenta líneas que te dan control total sobre la versión del cliente, las opciones de MQTT 5 y el estado de conexión. Ambos funcionan sobre WebSockets seguros (wss://), porque los navegadores no pueden abrir conexiones TCP.

Este tutorial muestra las dos opciones, conecta el resultado a un componente standalone con signals y lo prueba contra un broker real.

Ajustes de conexión

Ajuste	Valor
URL	`wss://broker.connectmq.com:8084/mqtt`
Host / puerto / ruta	`broker.connectmq.com` / `8084` / `/mqtt`
Protocolo	MQTT 3.1.1 o 5
Credenciales	una cuenta de navegador con acceso solo a los topics de la UI

Una app Angular servida por HTTPS debe usar wss://; ws:// se bloquea como contenido mixto. MQTT sobre WebSockets explica el transporte en detalle.

Opción 1: ngx-mqtt

Instalación

npm i ngx-mqtt

ngx-mqtt se mantiene de forma independiente de Angular y fija su propia dependencia de MQTT.js. Antes de instalarlo, comprueba que sus peer dependencies coinciden con tu versión de Angular; es la causa más habitual de errores de instalación.

Regístralo una vez en la configuración de una app standalone:

src/app/app.config.ts

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

Después inyecta MqttService donde necesites mensajes:

src/app/status.component.ts

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

observe() comparte una única suscripción en el broker entre todos los observadores del mismo filtro y la cancela cuando se va el último. publish() devuelve un observable que se completa cuando el broker confirma el mensaje; unsafePublish() dispara y se olvida.

Opción 2: tu propio servicio con MQTT.js

Si quieres la versión actual de MQTT.js, propiedades de MQTT 5 o simplemente menos dependencias, un servicio escrito a mano es sencillo:

Instalación

npm i mqtt

src/app/mqtt-client.service.ts

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; // sin MQTT durante 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);
    }
  }
}

Las decisiones de diseño:

Un cliente por app. providedIn: 'root' convierte el servicio en un singleton, así que todos los componentes comparten un WebSocket.
Observables fríos con conteo de referencias. No se suscribe nada en el broker hasta que un componente se suscribe, y el topic se cancela cuando lo hace el último componente. MQTT.js vuelve a suscribir los topics activos tras una reconexión por sí solo.
Seguro para SSR. En el servidor el servicio se queda en idle, así que Angular SSR o el prerendering nunca abren un socket.
Los wildcards se resuelven en el cliente. El broker envía cada mensaje una sola vez aunque coincidan varios filtros, así que matches() lo dirige a los observadores correctos. Prueba tus filtros en el topic matcher.

Un componente con signals

src/app/sensors.component.ts

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

toSignal() se suscribe en el contexto de inyección del componente y cancela la suscripción automáticamente cuando el componente se destruye: sin ngOnDestroy ni takeUntil. scan guarda el último valor por dispositivo, y auditTime(100) limita las actualizaciones de la UI a diez por segundo por muy rápido que lleguen los mensajes, lo que mantiene barata la detección de cambios en topics con mucho tráfico. (En versiones de Angular anteriores a la 19, añade standalone: true al decorador.)

Si prefieres observables en las plantillas, observe() funciona igual de bien con el pipe async o con takeUntilDestroyed() en un servicio.

Pruebas

Comprobación manual de punta a punta

Ejecuta ng serve y confirma que el componente muestra Broker: connected. Si muestra error, abre la consola del navegador: Connection refused: Not authorized indica credenciales o ACL; un handshake WebSocket fallido indica la URL, el puerto 8084 o la ruta /mqtt.
Abre el cliente MQTT online, conéctate al mismo broker y publica {"t": 23.4} en testmqtt/angular/sensors/room-1. La fila aparece en menos de 100 ms.
Publica en testmqtt/angular/sensors/room-2 y comprueba que se muestran ambas filas.
Suscríbete a testmqtt/angular/# en el cliente online y pulsa Ping para verificar la publicación.
Navega fuera del componente y confirma, con un debugger o en el log del broker, que se cancela la suscripción al topic.

Publica lecturas de prueba en tu app Angular

Tests unitarios

Los componentes no deberían necesitar un broker. Sustituye el servicio por un stub respaldado por un Subject:

src/app/sensors.component.spec.ts

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)); // espera a auditTime
  fixture.detectChanges();
  expect(fixture.nativeElement.textContent).toContain('room-1');
});

Prueba el servicio en sí, incluidos matches() y el comportamiento de reconexión, contra un broker real en una pequeña suite de integración. La guía de testing con Jest muestra patrones que se trasladan a los test runners de Angular, y MQTT con TypeScript cubre el tipado de payloads de punta a punta.

Solución de problemas

Síntoma	Causa	Solución
`npm i ngx-mqtt` falla con un conflicto de peer dependencies	La librería aún no está actualizada para tu versión de Angular	Usa el servicio MQTT.js de la opción 2
El build SSR se cuelga o agota el tiempo	El cliente se conecta en el servidor	Protégelo con `isPlatformBrowser` o `afterNextRender()`
La UI va lenta con topics muy activos	Detección de cambios en cada mensaje	Limita con `auditTime` o `throttleTime`
Las pestañas se desconectan entre sí	El mismo `clientId` en todas partes	Genera un client ID aleatorio por pestaña

Para más mensajes de error, consulta errores de MQTT.js.

Angular y MQTT encajan de forma natural porque ambos piensan en streams: un servicio singleton es dueño de la conexión, los componentes consumen observables o signals y la limpieza ocurre cuando el componente desaparece. Empieza con ngx-mqtt si soporta tu versión de Angular, pasa al servicio MQTT.js cuando necesites más control y verifica cada topic con un segundo cliente antes de publicar.

Preguntas frecuentes

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

ngx-mqtt es el arranque más rápido: una llamada a forRoot() y un método observe() que devuelve observables RxJS. Un pequeño servicio sobre MQTT.js te da la última versión de MQTT.js, opciones de MQTT 5 y control total, sin depender de que un wrapper siga el ritmo de Angular.

¿Tengo que cancelar la suscripción a los observables MQTT en los componentes Angular?

Sí, o usar algo que lo haga por ti. toSignal(), el pipe async y takeUntilDestroyed() cancelan la suscripción al destruirse el componente. Con un observe() con conteo de referencias, además se cancela la suscripción al topic en el broker cuando nadie escucha.

¿Por qué Angular SSR se cuelga o falla cuando mi app usa MQTT?

El cliente MQTT intenta abrir un WebSocket en el servidor e impide que el render termine. Conéctate solo en el navegador: comprueba isPlatformBrowser(PLATFORM_ID) antes de conectar o inicia la conexión en afterNextRender().

MQTT en Angular con ngx-mqtt y MQTT.js

Ajustes de conexión

Opción 1: ngx-mqtt

Opción 2: tu propio servicio con MQTT.js

Un componente con signals

Pruebas

Solución de problemas

Preguntas frecuentes

Sigue leyendo

MQTT en JavaScript: tutorial de MQTT.js para Node.js y el navegador

MQTT con tipado seguro en TypeScript

MQTT en React: dashboards en tiempo real con hooks