There are two good ways to use MQTT in Angular. ngx-mqtt wraps MQTT.js in an Angular module and gives you MqttService.observe(), which returns RxJS observables — the fastest way to get messages into a component. Or you write a small injectable service around MQTT.js yourself, which takes about sixty lines and gives you full control over the client version, MQTT 5 options and connection state. Both run over secure WebSockets (wss://), because browsers cannot open raw TCP connections.
This tutorial shows both, then wires the result into a standalone component with signals and tests it against a live broker.
Connection settings
| Setting | Value |
|---|---|
| URL | wss://broker.connectmq.com:8084/mqtt |
| Host / port / path | broker.connectmq.com / 8084 / /mqtt |
| Protocol | MQTT 3.1.1 or 5 |
| Credentials | a browser account with access only to the UI's topics |
An Angular app served over HTTPS must use wss://; ws:// is blocked as mixed content. MQTT over WebSockets explains the transport in detail.
Option 1: ngx-mqtt
npm i ngx-mqttngx-mqtt is maintained independently from Angular and pins its own MQTT.js dependency. Before installing, check that its peer dependencies match your Angular version; that is the most common source of install errors.
Register it once in the application config of a standalone app:
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))],
};Then inject MqttService wherever you need messages:
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() shares one broker subscription between all observers of the same filter and unsubscribes when the last one goes away. publish() returns an observable that completes when the broker acknowledges the message; unsafePublish() fires and forgets.
Option 2: your own MQTT.js service
If you want the current MQTT.js release, MQTT 5 properties, or simply fewer dependencies, a hand-written service is straightforward:
npm i mqttimport { 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; // no MQTT during 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);
}
}
}The design choices:
- One client per app.
providedIn: 'root'makes the service a singleton, so every component shares one WebSocket. - Cold, ref-counted observables. Nothing is subscribed on the broker until a component subscribes, and the topic is unsubscribed when the last component unsubscribes. MQTT.js re-subscribes active topics after a reconnect on its own.
- SSR-safe. On the server the service stays
idle, so Angular SSR or prerendering never opens a socket. - Wildcards are matched client-side. The broker sends each message once even if several filters match, so
matches()routes it to the right observers. Try your filters in the topic matcher.
A component with signals
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() subscribes in the component's injection context and unsubscribes automatically when the component is destroyed — no ngOnDestroy, no takeUntil. scan keeps the latest value per device, and auditTime(100) limits UI updates to ten per second however fast messages arrive, which keeps change detection cheap on busy topics. (On Angular versions before 19, add standalone: true to the decorator.)
If you prefer observables in templates, observe() works just as well with the async pipe or with takeUntilDestroyed() in a service.
Testing
Manual end-to-end check
- Run
ng serveand confirm the component showsBroker: connected. If it showserror, open the browser console:Connection refused: Not authorizedmeans credentials or ACLs; a failed WebSocket handshake means URL, port8084or path/mqtt. - Open the online MQTT client, connect to the same broker and publish
{"t": 23.4}totestmqtt/angular/sensors/room-1. The row appears within 100 ms. - Publish to
testmqtt/angular/sensors/room-2and check that both rows are shown. - Subscribe to
testmqtt/angular/#in the online client and click Ping to verify publishing. - Route away from the component and, in a debugger or the broker log, confirm the topic is unsubscribed.
Publish test readings to your Angular app
Unit tests
Components should not need a broker. Replace the service with a stub backed by a Subject:
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)); // wait for auditTime
fixture.detectChanges();
expect(fixture.nativeElement.textContent).toContain('room-1');
});Test the service itself, including matches() and reconnect behavior, against a real broker in a small integration suite. The Jest testing guide shows patterns that carry over to Angular test runners, and MQTT with TypeScript covers typing payloads end to end.
Troubleshooting
| Symptom | Cause | Fix |
|---|---|---|
npm i ngx-mqtt fails with a peer dependency conflict | Library not yet updated for your Angular version | Use the MQTT.js service from option 2 |
| SSR build hangs or times out | Client connects on the server | Guard with isPlatformBrowser or afterNextRender() |
| UI lags on busy topics | Change detection on every message | Throttle with auditTime or throttleTime |
| Tabs disconnect each other | Same clientId everywhere | Generate a random client ID per tab |
For more error strings, see MQTT.js errors.
Angular and MQTT fit together naturally because both think in streams: a singleton service owns the connection, components consume observables or signals, and cleanup happens when the component goes away. Start with ngx-mqtt if it supports your Angular version, switch to the MQTT.js service when you need more control, and verify every topic with a second client before you ship.
Frequently asked questions
Should I use ngx-mqtt or MQTT.js directly in Angular?
ngx-mqtt is the quickest start: one forRoot() call and an observe() method that returns RxJS observables. A small service around MQTT.js gives you the latest MQTT.js, MQTT 5 options and full control, and does not depend on a wrapper keeping pace with Angular releases.
Do I need to unsubscribe from MQTT observables in Angular components?
Yes, or use something that does it for you. toSignal(), the async pipe and takeUntilDestroyed() all unsubscribe when the component is destroyed. With a ref-counted observe() method that also unsubscribes the topic on the broker once nobody listens.
Why does Angular SSR hang or error when my app uses MQTT?
The MQTT client tries to open a WebSocket on the server and keeps the render from finishing. Only connect in the browser: check isPlatformBrowser(PLATFORM_ID) before connecting, or start the connection in afterNextRender().
