Angular MQTT Tutorial with RxJS, MQTT.js and ngx-mqtt

Angular and MQTT fit together naturally: MQTT pushes messages, and RxJS observables are Angular's native way to handle pushed data. In this Angular MQTT tutorial you will build an injectable service that connects to a broker over secure WebSocket, exposes any topic filter as an observable, and publishes commands, then render live values with the async pipe. You will also see how the popular ngx-mqtt library compares, so you can choose the right approach for your project.

Architecture: One Connection, Many Observables

The core idea is simple. A single root-level service owns the MQTT.js client. Components ask the service for a topic filter and get back an Observable. When the first component subscribes to a filter, the service sends an MQTT SUBSCRIBE; when the last one unsubscribes, the service sends UNSUBSCRIBE.

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

This gives you exactly one WebSocket per browser tab, automatic cleanup through RxJS teardown, and full support for + and # wildcards. Browsers cannot use raw TCP, so the service uses wss:// on port 8084 with the /mqtt path, as explained in MQTT over WebSocket.

Setup

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

The examples use standalone components and the inject() function, available in current Angular versions.

The MqttService

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

/** MQTT topic filter matching with + and # wildcards. */
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");
      // restore all active subscriptions after (re)connect
      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 {
        /* keep text */
      }
      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");
  }
}

How the Observable Wiring Works

• share() plus the stream cache means several components observing the same filter share one inner subscription. Two components that both observe plant/+/temp cause a single SUBSCRIBE packet, and one leaving does not cut off the other.
• finalize runs when the last subscriber leaves, which decrements the reference count and triggers the MQTT UNSUBSCRIBE, so the broker stops sending messages nobody displays.
• Wildcard matching happens client-side with topicMatches. The broker already filters by subscription, but overlapping filters (for example plant/# and plant/line1/temp) deliver messages through one stream, so each observable must check which messages belong to it.
• Change detection. MQTT.js callbacks fire from WebSocket events, which zone.js patches, so Angular detects changes automatically. In zoneless applications, use the async pipe or toSignal(), both of which mark the view for update.

Rendering Live Data with the Async Pipe

Create a component in 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);
  }
}

The scan operator folds the stream into a map of the latest message per topic, so the panel shows one card per device instead of a growing list. Then connect once at startup in 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");
  }
}

Run ng serve, then publish test data to the public MQTT test broker:

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

The card appears immediately. The public area is readable by anyone, so keep your own random prefix and only test data.

Signals, OnPush and Testing

Recent Angular versions push applications toward signals and OnPush change detection, and the observable service fits that direction without changes. Convert any stream at the edge of a component with toSignal(), for example readonly reading = toSignal(this.mqtt.observePayload<Telemetry>(topic), { initialValue: {} }), and read reading() in the template. The signal unsubscribes automatically when the component is destroyed, so the reference counting in the service still works.

For unit tests, avoid connecting to a real broker. Because components only depend on observe, observePayload, publish and state$, you can provide a fake service with { provide: MqttService, useValue: fake } where observe returns a Subject you control. Push messages into the subject and assert on the rendered output. Keep a small set of integration tests that connect to the public broker with a random prefix, which verifies the real wiring, the WebSocket URL and the topic structure end to end without slowing down every test run.

Retained messages deserve a mention here too. If devices publish their latest telemetry with the retain flag, a component that subscribes after navigation receives the current value straight away instead of showing empty placeholders until the next report. The retain field on MqttMessage lets you distinguish that initial snapshot from live updates, which is useful when you want to animate only fresh changes or mark the first value as possibly old.

Finally, keep topic strings out of components. A small constants file or a typed topic builder means a renamed topic is a one-line change, and it makes searching the codebase for every consumer of a topic trivial.

What About ngx-mqtt?

ngx-mqtt is a long-standing Angular wrapper around MQTT.js. It provides an MqttService with observe(filter) returning observables of IMqttMessage, configured once at 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))],
});

How the two approaches compare:

If ngx-mqtt's latest release supports your Angular version, it is a quick start. If you want current MQTT.js features or fewer upgrade constraints, the custom service is small enough to own outright.

Troubleshooting Angular MQTT Apps

1. global is not defined in the browser. An older MQTT.js or a dependency expects Node globals. Upgrade to MQTT.js v5, or as a stopgap add (window as any).global = window; at the top of main.ts.
2. Connection loops between tabs. A fixed client ID is shared. The service generates a random one unless you override it.
3. Messages stop after navigation. A component unsubscribed and was the last observer, so the service unsubscribed. That is by design; the next observer resubscribes.
4. No updates in a zoneless app. Bind with the async pipe or toSignal() instead of assigning values in a manual subscribe callback.

Production Setup with ConnectMQ

For production, create a private broker on ConnectMQ and pass credentials to connect(): connect("wss://YOUR_BROKER_HOST:8084/mqtt", { username, password, clientId }), using values from the portal or a token-issuing backend. Because frontend code is visible to users, use a dedicated dashboard credential and route sensitive commands through your backend. A cloud MQTT broker with a Let's Encrypt certificate on the secure WebSocket port works in every browser without extra setup.

Related Tutorials

The same ideas appear in the React MQTT hook and the Vue MQTT composable. To add typed topic contracts on top of this service, read type-safe MQTT in TypeScript, and for a broader overview of client options, see the MQTT client libraries guide. The RxJS documentation covers share, scan and finalize in detail.

Get Started with ConnectMQ

Connect your Angular app to a private broker with secure WebSocket today. Sign up at app.connectmq.com, choose the free Serverless plan with no credit card, create a broker and pass the generated credentials to the service. Use the portal's in-browser MQTT client to publish test messages while you build your components.

Frequently Asked Questions

Should I use ngx-mqtt or MQTT.js directly in Angular?

Use ngx-mqtt if its current release supports your Angular version and you only need basic observe and publish. Use MQTT.js directly with a small service when you want MQTT 5 features, the newest MQTT.js and full control over upgrades.

Can Angular connect to MQTT over TCP port 1883 or 8883?

No. Browsers only allow WebSocket connections, so Angular apps use wss:// on port 8084 with the /mqtt path on ConnectMQ brokers.

How do I convert MQTT messages to Angular signals?

Pass the observable to toSignal() from @angular/core/rxjs-interop, for example toSignal(mqtt.observePayload<Telemetry>(topic)). The signal updates whenever a new message arrives.

Does every component open its own MQTT connection?

Not with this design. The service is provided in root, so the whole application shares one connection, and each observed topic filter is subscribed once regardless of how many components use it.