Consumers and Subscriptions

Consumers and Subscriptions

A consumer is a named subscriber:

const consumer = await mq.createConsumer('dashboard');

// stable identity for durable delivery (defaults to the name):
const worker = await mq.createConsumer('worker', { consumer_id: 'worker-7' });

Consumer names are unique per realm: a second live socket registering an existing name is rejected with an error and disconnected (this protects you from two instances accidentally sharing one app_id).

Subscribe forms

// 1. A specific event from a specific producer
consumer.subscribe('weather-station', 'temperature', handler);

// 2. A specific event from ANY producer (note the options argument)
consumer.subscribe('temperature', handler, {});

// 3. ALL events from one producer
consumer.subscribe('weather-station', handler);

// 4. An MQTT-style topic pattern — see [[Topics and Wildcards]]
consumer.subscribe('org/acme/machine/+/cmd', handler, { mode: 'topic' });

All forms take an options object as the last argument before/after the callback (see below) and re-subscribe automatically on reconnect (subscribeOnce opts out of that).

The handler

consumer.subscribe(producer, event, function (message, from, ack, raw) {
    // message — the payload
    // from    — producing client's name
    // ack     — function; call it for manual ACK (see reliability page)
    // raw     — full envelope: { event, message, from, msgId?, delivery_attempt? }
});

raw.event is useful in topic mode — it carries the concrete topic that matched your pattern.

Subscription options

Option	Meaning
durable: true	queue messages while this consumer is offline — Persistence and Durable Delivery
consumer_id	stable identity for durable replay (defaults to the consumer's id/name)
ack: true	request ACK-tracked delivery (auto-ACK after the handler resolves)
manual_ack: true	ACK-tracked, but you call ack() yourself
ack_timeout: '30s'	how long the server waits for an ACK before retrying
retry: { max_attempts, delay, backoff }	retry policy — ACK, Retry and the Dead Letter Queue
mode: 'topic'	treat the event as an MQTT-style pattern — Topics and Wildcards
group: 'name'	join a consumer group — Consumer Groups
reconnect: false	do not re-subscribe automatically after reconnect

Connection state callbacks

Consumers learn when their producers come and go:

consumer.setOnProducerOnlineListener('weather-station', () => { ... });

The server emits per-client CONNECT-<id> / DISCONNECT-<id> events as producers/consumers go on- and offline; the client library wires these up for you.

Unsubscribing

consumer.unsubscribe('temperature', 'weather-station');  // one subscription
consumer.clearSubscriptions();                           // all consume handlers

Losing the connection does not unsubscribe — subscriptions are part of the server-side registry and survive disconnects by design.

Authentication

With auth enabled, the subscribing socket needs the consumer, both, manager, or admin role. How consumers get in (pre-shared realm keys, automatic admission) is covered in Roles and Connection Authorization.