MQTT

MqttActor integrates with MQTT brokers (Mosquitto, EMQ X, HiveMQ, AWS IoT Core). Supports both MQTT 3.1.1 and 5; topic wildcards (+, #); QoS 0/1/2; retained messages.

It is the MQTT counterpart to WebsocketClientActor: an abstract base class you extend. Declare subscriptions in the constructor, handle inbound traffic in onMessage, and publish with this.publish(...). It is still controllable from the outside via ref.tell(...).

import { ActorSystem, ActorSystemOptions } from 'actor-ts';
import { MqttActor, MqttOptions, type MqttMessage } from 'actor-ts/io';

type Reading = { sensor: string; celsius: number };

const actorSystemOptions = ActorSystemOptions.create().withConfig({
    'actor-ts': { io: { broker: { mqtt: { brokerUrl: 'mqtt://localhost:1883' } } } },
  });
class TemperatureHub extends MqttActor<Reading> {
  constructor(options: MqttOptions) {
    super(options.withQos(1).withClientId('temperature-hub'));
    this.subscribe('sensors/+/temp');
    this.subscribe('alerts/#', { qos: 2 });
  }

  override onMessage(message: MqttMessage<Reading>): void {
    const { sensor, celsius } = message.payload.entity();   // decoded via the codec
    this.log.info(`${sensor}: ${celsius}°C`);
    this.publish(`ack/${sensor}`, 'ok');                // raw string
    this.publish('rollup', { sensor, celsius });        // encoded entity
  }
}

const system = ActorSystem.create('demo', actorSystemOptions);

system.spawn(() => new TemperatureHub(MqttOptions.create()), 'hub');

T (here Reading) types the inbound payload — message.payload.entity() returns a T. A second generic, TSelf, types application messages other actors may tell the ref (see External control below); it defaults to never.

Hooks

Override these on your subclass. onReceive and onCommand are sealed — the base classes dispatch to the hooks below; do not override them.

Hook	When
onMessage(message)	A message arrived on one of this actor’s own subscriptions. Required.
onConnected()	The connection (re)opened; the subscription registry has been re-applied on the broker.
onDisconnected(cause?)	The connection dropped; a reconnect cycle may follow (per settings).
onInvalidMessage(err, message)	onMessage threw an MqttDecodeError (a lazy entity() on a malformed payload). Default: log + drop. Rethrow to escalate.
onSelfMessage(message)	An application message (TSelf) was told to this ref.
onTerminated(signal)	An actor you context.watched stopped. Inherited from BrokerActor; delivery targets this actor watches itself are already handled.

Inbound and lifecycle events are delivered through the mailbox, so onMessage and the hooks always run on the actor thread (single-threaded, per-connection order preserved: connected → messages → disconnected).

Configuration

Settings resolve with the usual precedence: constructor / builder > HOCON (actor-ts.io.broker.mqtt) > built-in defaults. Use the fluent MqttOptions builder, or pass a plain Partial<MqttOptionsType>.

const options = MqttOptions.create()
  .withBrokerUrl('mqtts://mqtt.example.com:8883')
  .withClientId('my-app')
  .withCredentials(process.env.MQTT_USER, process.env.MQTT_PASS)
  .withQos(1)                 // default QoS for publish/subscribe
  .withProtocolVersion(5)     // opt in to MQTT 5.0
  .withCleanSession(false)
  .withKeepAlive(30);

interface MqttOptionsType extends BrokerCommonOptionsType {
  brokerUrl?:  string;          // mqtt:// mqtts:// ws:// wss://
  clientId?:   string;
  credentials?: { username?: string; password?: string };
  qos?: 0 | 1 | 2;              // default QoS for publish/subscribe
  cleanSession?: boolean;       // default true
  keepAlive?: number;           // seconds; default 60
  protocolVersion?: 4 | 5;      // default 4 (= MQTT 3.1.1)
  codec?: MqttCodec<unknown>;   // default mqttJsonCodec()
  will?: { topic: string; payload: string | Uint8Array; qos?: 0 | 1 | 2; retain?: boolean };
  tls?: TlsTransportOptionsType;  // cert material for mqtts:// / wss://; code-only
}

Common patterns:

• withCleanSession(false) — session persists; the broker delivers messages missed during disconnection (subject to broker config).
• withWill({ ... }) — the broker publishes this when the client disconnects ungracefully. Useful for presence (“device-42-offline”).
• withProtocolVersion(5) — enables MQTT 5 features (user properties, reason codes on MqttMessage). Requires a v5-capable broker.

Typed payloads

Inbound MqttMessage<T> carries a lazily-decoding MqttPayload<T>:

override onMessage(message: MqttMessage<Reading>): void {
  message.payload.bytes;          // raw Uint8Array
  message.payload.text();         // UTF-8 string (cached)
  message.payload.entity();       // decoded T via the codec (cached)
  message.payload.entity<Acknowledgment>();  // decode as a different type on a specific topic
}

Decoding is lazy — it only happens when you call text() / entity(). A malformed payload makes entity() throw an MqttDecodeError; because that happens inside onMessage, the base class catches it and routes it to onInvalidMessage (default: log + drop, no restart):

protected override onInvalidMessage(err: MqttDecodeError, message: MqttMessage<Reading>): void {
  this.log.warn(`bad payload on ${err.topic}: ${err.message}`);
}

Publishing

this.publish(topic, payload, options?) returns false if the message was dropped (encode failure or outbound-buffer overflow):

• a string or Uint8Array payload is sent raw;
• any other value is encoded via the codec (JSON by default).

this.publish('ack/1', 'ok');                    // raw bytes: ok
this.publish('rollup', { sensor, celsius });    // JSON: {"sensor":...,"celsius":...}
this.publish('cfg', data, { qos: 1, retain: true });

To publish a bare string as a JSON entity (the wire bytes "pong" rather than pong), encode it explicitly:

this.publish('topic', this.codec().encode('pong'));

Codecs

The codec turns entities into bytes and back. The default is mqttJsonCodec() (plain JSON over UTF-8). Supply your own via withCodec(...) — e.g. a JSON codec with runtime validation:

import { mqttJsonCodec, MqttOptions } from 'actor-ts/io';

const codec = mqttJsonCodec<Reading>({
  validate: (v) => ReadingSchema.parse(v),   // zod, etc.; throws → MqttDecodeError
});

MqttOptions.create().withBrokerUrl('mqtt://localhost:1883').withCodec(codec);

External control

Beyond the subclass API, any actor can drive an MqttActor by telling it an MqttCommand:

ref.tell({ kind: 'publish', publish: { topic: 't', payload: 'hi', qos: 1 } });

// Subscribe and route to this actor's own onMessage (no target):
ref.tell({ kind: 'subscribe', topic: 'x/#', qos: 1 });

// Subscribe and fan out to another actor (external target):
ref.tell({ kind: 'subscribe', topic: 'y/#', target: someHandler });

ref.tell({ kind: 'unsubscribe', topic: 'y/#', target: someHandler });

• A subscribe with no target delivers matching messages to the actor’s own onMessage; with a target, it fans out to that actor. Overlapping patterns deliver to each ref at most once.
• An unsubscribe with a target removes that target. With no target, it removes all foreign targets but leaves the actor’s own subscription intact — an external controller can’t silence the subclass’s constructor-declared subscription.
• Fan-out targets are deathwatched: when a target actor stops, it is pruned automatically (and a broker UNSUBSCRIBE fires once a pattern has no consumers left).

Since these commands are plain objects, keep any TSelf you define off the kind values publish / subscribe / unsubscribe — otherwise it would be dispatched as a command instead of reaching onSelfMessage.

QoS levels

QoS	Delivery	Cost
0	At-most-once. Fire-and-forget.	Cheapest; messages may be lost.
1	At-least-once. Broker stores until acked.	Default for most use cases. Duplicates possible.
2	Exactly-once. Full handshake.	Slowest; rarely needed.

For IoT telemetry, QoS 1 is the typical balance. Commands to devices: also QoS 1 (or QoS 2 if duplicates would be harmful). Status updates that are quickly superseded (sensor readings): QoS 0 is fine.

Topic wildcards

this.subscribe('sensors/+/temp');   // matches sensors/dev1/temp, sensors/dev2/temp, ...
this.subscribe('devices/#');        // matches devices/anything/anywhere/...

Wildcard	Matches
+	Exactly one level.
#	Multiple levels (must be the last segment).

Retained messages

this.publish('config/device-42', { mode: 'eco' }, { retain: true });

A retained message is stored by the broker and delivered immediately to any new subscriber. Use for configuration (devices reading the latest config on connect) or last-known-state. Keep retain unset for normal pub/sub — otherwise the broker keeps every old message visible forever.

TLS

An mqtts:// or wss:// URL starts the handshake, and mqtt.js verifies the broker against the system trust store by default. Behind a private CA, or for client-certificate auth, add withTls:

const mqttOptions = MqttOptions.create()
  .withBrokerUrl('mqtts://mqtt.example.com:8883')
  .withTls({
    ca: fs.readFileSync('./tls/ca.crt', 'utf8'),
    cert: fs.readFileSync('./tls/device.crt', 'utf8'),
    key: fs.readFileSync('./tls/device.key', 'utf8'),
  });

Every field carries the material itself, never a path to it, which is why the example reads the files. cert and key only mean anything together, and one without the other is rejected when the actor starts.

withTls configures the handshake rather than starting one, so keep the mqtts:// / wss:// scheme. It has no config key on purpose: a private key does not belong in application.conf.

Peer dependency

npm install mqtt
# or: bun add mqtt

When to use MQTT

1. IoT / telemetry — devices publishing sensor data, subscribing to commands.
2. Bridging from existing MQTT infrastructure — when the broker is already there.
3. Lightweight pub/sub — when Kafka is overkill (small payloads, no replay history needed).

For cluster-internal pub/sub, DistributedPubSub is simpler (no external broker).

Migrating from the old MqttActor

MqttActor used to be a concrete class you spawned and drove only with tell. It is now abstract — extend it and override onMessage.

• new MqttActor({...}) → class X extends MqttActor<T> { onMessage(m) {…} }.
• message.payload (a Uint8Array) → message.payload.bytes; new TextDecoder().decode(message.payload) → message.payload.text(); JSON → message.payload.entity().
• settings.subscriptions: [{ topic, target }] (removed) → this.subscribe(topic, { target }) in the constructor, or a subscribe command.

Persistent session and `cleanSession: false`

Persistent sessions need the same clientId across reconnects — the broker tracks subscriptions + queued messages per-clientId. A random clientId each connect makes cleanSession: false useless.

Wildcard subscriptions multiply traffic

# at the root subscribes to every message on the broker. In a multi-tenant broker, this is rarely what you want. Scope wildcards narrowly.

Where to next

• I/O overview — the bigger picture.
• BrokerActor base — the shared lifecycle.
• WebSocket client — the same subclass-first shape.
• Kafka — for durable streaming.