Tutorial de MQTT com Angular usando RxJS, MQTT.js e ngx-mqtt

Angular e MQTT combinam naturalmente: o MQTT envia mensagens por push, e os observables do RxJS são a forma nativa do Angular de lidar com dados enviados por push. Neste tutorial de MQTT com Angular você vai construir um serviço injetável que se conecta a um broker via WebSocket seguro, expõe qualquer filtro de topic como um observable e publica comandos, e depois renderizar valores ao vivo com o pipe async, por exemplo no painel de uma linha de produção em uma planta de Joinville. Você também verá como a popular biblioteca ngx-mqtt se compara, para escolher a abordagem certa para o seu projeto.

Arquitetura: Uma Conexão, Muitos Observables

A ideia central é simples. Um único serviço no nível raiz é dono do cliente MQTT.js. Os componentes pedem ao serviço um filtro de topic e recebem de volta um Observable. Quando o primeiro componente assina um filtro, o serviço envia um SUBSCRIBE MQTT; quando o último cancela a assinatura, o serviço envia um UNSUBSCRIBE.

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

Isso dá exatamente um WebSocket por aba do navegador, limpeza automática pelo teardown do RxJS e suporte completo aos curingas + e #. Navegadores não podem usar TCP bruto, então o serviço usa wss:// na porta 8084 com o caminho /mqtt, como explicado em MQTT sobre WebSocket.

Configuração

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

Os exemplos usam componentes standalone e a função inject(), disponíveis nas versões atuais do Angular.

O MqttService

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

/** Correspondência de filtros de topic MQTT com os curingas + e #. */
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 as assinaturas ativas após a (re)conexão
      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 {
        /* mantém o 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");
  }
}

Como Funciona a Ligação dos Observables

• share() junto com o cache de streams faz com que vários componentes que observam o mesmo filtro compartilhem uma única assinatura interna. Dois componentes que observam plant/+/temp geram um único pacote SUBSCRIBE, e a saída de um não corta o outro.
• finalize é executado quando o último assinante sai, o que decrementa a contagem de referências e dispara o UNSUBSCRIBE MQTT, para que o broker pare de enviar mensagens que ninguém exibe.
• A correspondência de curingas acontece no cliente com topicMatches. O broker já filtra pela assinatura, mas filtros sobrepostos (por exemplo plant/# e plant/line1/temp) entregam mensagens por um único fluxo, então cada observable precisa verificar quais mensagens lhe pertencem.
• Detecção de mudanças. Os callbacks do MQTT.js são disparados por eventos do WebSocket, que o zone.js intercepta, então o Angular detecta as mudanças automaticamente. Em aplicações zoneless, use o pipe async ou toSignal(), que marcam a view para atualização.

Renderizando Dados ao Vivo com o Pipe Async

Crie um componente em 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);
  }
}

O operador scan condensa o fluxo em um mapa com a última mensagem de cada topic, então o painel mostra um card por dispositivo em vez de uma lista crescente. Depois, conecte uma única vez na inicialização em 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");
  }
}

Execute ng serve e depois publique dados de teste no servidor MQTT público (gratuito), que serve apenas para testes porque qualquer pessoa pode ler seus topics:

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

O card aparece imediatamente. O servidor MQTT público pode ser lido por qualquer pessoa, então mantenha seu próprio prefixo aleatório e use apenas dados de teste.

Signals, OnPush e Testes

As versões recentes do Angular empurram as aplicações para signals e para a detecção de mudanças OnPush, e o serviço baseado em observables se encaixa nessa direção sem alterações. Converta qualquer fluxo na borda de um componente com toSignal(), por exemplo readonly reading = toSignal(this.mqtt.observePayload<Telemetry>(topic), { initialValue: {} }), e leia reading() no template. O signal cancela a assinatura automaticamente quando o componente é destruído, então a contagem de referências do serviço continua funcionando.

Nos testes unitários, evite se conectar a um broker real. Como os componentes dependem apenas de observe, observePayload, publish e state$, você pode fornecer um serviço falso com { provide: MqttService, useValue: fake }, em que observe retorna um Subject que você controla. Envie mensagens para o subject e verifique a saída renderizada. Mantenha um pequeno conjunto de testes de integração que se conectam ao servidor MQTT público com um prefixo aleatório, o que verifica de ponta a ponta a ligação real, a URL do WebSocket e a estrutura de topics, sem deixar lenta cada execução dos testes.

As retained messages também merecem uma menção aqui. Se os dispositivos publicam sua telemetria mais recente com a flag retain, um componente que assina após uma navegação recebe o valor atual imediatamente, em vez de mostrar placeholders vazios até o próximo relatório. O campo retain em MqttMessage permite distinguir esse snapshot inicial das atualizações ao vivo, o que é útil quando você quer animar apenas as mudanças recentes ou marcar o primeiro valor como possivelmente antigo.

Por fim, mantenha as strings de topic fora dos componentes. Um pequeno arquivo de constantes ou um construtor de topics tipado faz com que renomear um topic seja uma mudança de uma linha, e torna trivial procurar no código todos os consumidores de um topic.

E o ngx-mqtt?

O ngx-mqtt é um wrapper Angular de longa data em torno do MQTT.js. Ele fornece um MqttService com observe(filter) retornando observables de IMqttMessage, configurado uma vez no 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))],
});

Como as duas abordagens se comparam:

Se a versão mais recente do ngx-mqtt suporta a sua versão do Angular, ela é um começo rápido. Se você quer os recursos atuais do MQTT.js ou menos restrições nas atualizações, o serviço próprio é pequeno o bastante para ficar inteiramente sob seu controle.

Solução de Problemas em Apps de MQTT com Angular

1. global is not defined no navegador. Um MQTT.js antigo ou alguma dependência espera globais do Node. Atualize para o MQTT.js v5 ou, como paliativo, adicione (window as any).global = window; no início de main.ts.
2. Loops de conexão entre abas. Um client ID fixo está sendo compartilhado. O serviço gera um aleatório, a menos que você o substitua.
3. As mensagens param após uma navegação. Um componente cancelou a assinatura e era o último observador, então o serviço cancelou a assinatura MQTT. Isso é proposital; o próximo observador assina de novo.
4. Nenhuma atualização em um app zoneless. Faça o binding com o pipe async ou toSignal(), em vez de atribuir valores em um callback subscribe manual.

Configuração para Produção com o ConnectMQ

Em produção, crie um broker privado no ConnectMQ, um broker MQTT gerenciado, e passe as credenciais para connect(): connect("wss://YOUR_BROKER_HOST:8084/mqtt", { username, password, clientId }), usando valores do portal ou de um backend que emite tokens. Como o código do frontend é visível para os usuários, use uma credencial dedicada para o dashboard e encaminhe comandos sensíveis pelo seu backend. Um broker MQTT na nuvem com um certificado Let's Encrypt na porta WebSocket segura funciona em todos os navegadores sem configuração extra.

Tutoriais Relacionados

As mesmas ideias aparecem no hook de MQTT com React e no composable de MQTT com Vue. Para adicionar contratos de topic tipados sobre este serviço, leia MQTT com tipagem segura em TypeScript e, para uma visão mais ampla das opções de cliente, veja o guia de bibliotecas cliente MQTT. A documentação do RxJS cobre share, scan e finalize em detalhes.

Comece a Usar o ConnectMQ

Conecte hoje mesmo seu app Angular a um broker privado com WebSocket seguro. Cadastre-se em app.connectmq.com, escolha o plano Serverless gratuito, sem cartão de crédito, crie um broker e passe as credenciais geradas para o serviço. Use o cliente MQTT no navegador do portal para publicar mensagens de teste enquanto constrói seus componentes.

Perguntas frequentes

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

Use o ngx-mqtt se a versão atual dele suporta a sua versão do Angular e você só precisa de observe e publish básicos. Use o MQTT.js diretamente com um pequeno serviço quando quiser recursos do MQTT 5, o MQTT.js mais recente e controle total sobre as atualizações.

O Angular pode se conectar ao MQTT pelas portas TCP 1883 ou 8883?

Não. Os navegadores só permitem conexões WebSocket, então apps Angular usam wss:// na porta 8084 com o caminho /mqtt nos brokers do ConnectMQ.

Como converto mensagens MQTT em signals do Angular?

Passe o observable para toSignal() de @angular/core/rxjs-interop, por exemplo toSignal(mqtt.observePayload<Telemetry>(topic)). O signal é atualizado sempre que chega uma nova mensagem.

Cada componente abre sua própria conexão MQTT?

Não com este design. O serviço é fornecido na raiz, então a aplicação inteira compartilha uma única conexão, e cada filtro de topic observado é assinado uma única vez, independentemente de quantos componentes o usem.