MQTT no Angular: ngx-mqtt e um serviço MQTT.js com RxJS

Existem duas boas formas de usar MQTT no Angular. O ngx-mqtt envolve o MQTT.js em um módulo Angular e oferece MqttService.observe(), que retorna observables RxJS: o caminho mais rápido para levar mensagens a um componente. Ou você escreve um pequeno serviço injetável sobre o MQTT.js, cerca de sessenta linhas que dão controle total sobre a versão do cliente, as opções do MQTT 5 e o estado da conexão. Os dois rodam sobre WebSockets seguros (wss://), porque navegadores não abrem conexões TCP.

Este tutorial mostra as duas opções, liga o resultado a um componente standalone com signals e testa tudo contra um broker real.

Configurações de conexão

Configuração	Valor
URL	`wss://broker.connectmq.com:8084/mqtt`
Host / porta / path	`broker.connectmq.com` / `8084` / `/mqtt`
Protocolo	MQTT 3.1.1 ou 5
Credenciais	uma conta de navegador com acesso só aos topics da UI

Uma app Angular servida por HTTPS precisa usar wss://; ws:// é bloqueado como conteúdo misto. MQTT sobre WebSockets explica o transporte em detalhes.

Opção 1: ngx-mqtt

Instalação

npm i ngx-mqtt

O ngx-mqtt é mantido de forma independente do Angular e fixa a própria dependência do MQTT.js. Antes de instalar, confira se as peer dependencies batem com a sua versão do Angular; essa é a causa mais comum de erros de instalação.

Registre-o uma vez na configuração de uma 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))],
};

Depois injete o MqttService onde precisar de mensagens:

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() compartilha uma única assinatura no broker entre todos os observadores do mesmo filtro e a cancela quando o último sai. publish() retorna um observable que completa quando o broker confirma a mensagem; unsafePublish() dispara e esquece.

Opção 2: seu próprio serviço com MQTT.js

Se você quer a versão atual do MQTT.js, propriedades do MQTT 5 ou simplesmente menos dependências, um serviço escrito à mão é simples:

Instalação

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

As escolhas de design:

Um cliente por app. providedIn: 'root' transforma o serviço em singleton, então todos os componentes compartilham um WebSocket.
Observables frios com contagem de referências. Nada é assinado no broker até um componente se inscrever, e o topic é cancelado quando o último componente sai. O MQTT.js reassina os topics ativos sozinho após uma reconexão.
Seguro para SSR. No servidor o serviço fica em idle, então o Angular SSR ou o prerendering nunca abrem um socket.
Wildcards são resolvidos no cliente. O broker envia cada mensagem uma vez, mesmo que vários filtros casem, então matches() a encaminha aos observadores certos. Teste seus filtros no topic matcher.

Um componente com 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 inscreve no contexto de injeção do componente e cancela a inscrição automaticamente quando o componente é destruído: sem ngOnDestroy, sem takeUntil. scan guarda o último valor por dispositivo, e auditTime(100) limita as atualizações da UI a dez por segundo, não importa quão rápido as mensagens cheguem, o que mantém a detecção de mudanças barata em topics movimentados. (Em versões do Angular anteriores à 19, adicione standalone: true ao decorator.)

Se preferir observables nos templates, observe() funciona igualmente bem com o pipe async ou com takeUntilDestroyed() em um serviço.

Testes

Verificação manual de ponta a ponta

Rode ng serve e confirme que o componente mostra Broker: connected. Se mostrar error, abra o console do navegador: Connection refused: Not authorized indica credenciais ou ACLs; um handshake WebSocket com falha indica a URL, a porta 8084 ou o path /mqtt.
Abra o cliente MQTT online, conecte ao mesmo broker e publique {"t": 23.4} em testmqtt/angular/sensors/room-1. A linha aparece em até 100 ms.
Publique em testmqtt/angular/sensors/room-2 e confira que as duas linhas aparecem.
Assine testmqtt/angular/# no cliente online e clique em Ping para verificar a publicação.
Navegue para fora do componente e confirme, com um debugger ou no log do broker, que a assinatura do topic foi cancelada.

Publique leituras de teste na sua app Angular

Testes unitários

Componentes não deveriam precisar de um broker. Substitua o serviço por um stub baseado em um 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)); // aguarda o auditTime
  fixture.detectChanges();
  expect(fixture.nativeElement.textContent).toContain('room-1');
});

Teste o serviço em si, incluindo matches() e o comportamento de reconexão, contra um broker real em uma pequena suíte de integração. O guia de testes com Jest mostra padrões que valem também para os test runners do Angular, e MQTT com TypeScript cobre a tipagem de payloads de ponta a ponta.

Solução de problemas

Sintoma	Causa	Correção
`npm i ngx-mqtt` falha com conflito de peer dependency	A biblioteca ainda não foi atualizada para a sua versão do Angular	Use o serviço MQTT.js da opção 2
O build SSR trava ou estoura o tempo	O cliente conecta no servidor	Proteja com `isPlatformBrowser` ou `afterNextRender()`
A UI fica lenta em topics movimentados	Detecção de mudanças a cada mensagem	Limite com `auditTime` ou `throttleTime`
Abas derrubam umas às outras	Mesmo `clientId` em todo lugar	Gere um client ID aleatório por aba

Para mais mensagens de erro, veja erros do MQTT.js.

Angular e MQTT combinam naturalmente porque os dois pensam em streams: um serviço singleton é dono da conexão, os componentes consomem observables ou signals e a limpeza acontece quando o componente sai de cena. Comece com o ngx-mqtt se ele suportar a sua versão do Angular, passe para o serviço MQTT.js quando precisar de mais controle e verifique cada topic com um segundo cliente antes de publicar.

Perguntas frequentes

Devo usar ngx-mqtt ou MQTT.js direto no Angular?

O ngx-mqtt é o início mais rápido: uma chamada a forRoot() e um método observe() que retorna observables RxJS. Um pequeno serviço sobre o MQTT.js dá a versão mais recente do MQTT.js, opções do MQTT 5 e controle total, sem depender de um wrapper acompanhar o ritmo do Angular.

Preciso cancelar a inscrição dos observables MQTT nos componentes Angular?

Sim, ou usar algo que faça isso por você. toSignal(), o pipe async e takeUntilDestroyed() cancelam a inscrição quando o componente é destruído. Com um observe() com contagem de referências, a assinatura do topic no broker também é cancelada quando ninguém mais escuta.

Por que o Angular SSR trava ou dá erro quando minha app usa MQTT?

O cliente MQTT tenta abrir um WebSocket no servidor e impede que o render termine. Conecte apenas no navegador: verifique isPlatformBrowser(PLATFORM_ID) antes de conectar ou inicie a conexão em afterNextRender().

MQTT no Angular com ngx-mqtt e MQTT.js

Configurações de conexão

Opção 1: ngx-mqtt

Opção 2: seu próprio serviço com MQTT.js

Um componente com signals

Testes

Solução de problemas

Perguntas frequentes

Continue lendo

MQTT em JavaScript: tutorial de MQTT.js para Node.js e navegador

MQTT com tipagem segura em TypeScript

MQTT no React: dashboards em tempo real com hooks