Getting started

Creating a relay

Create one Relay per process. It uses the same connection options as queues and workers, and it is safe to share it between all the code of that process.

Redis

import { Relay } from 'bullmq';

const relay = new Relay({
  connection: { host: 'localhost', port: 6379 },
});

await relay.waitUntilReady();

PostgreSQL

import { Relay, createPostgresRelayBackend } from 'bullmq';

const relay = new Relay({
  connection: {
    connectionString: 'postgres://user:password@localhost:5432/app',
    migrate: true, // or run the migrations yourself, see below
  },
  backend: createPostgresRelayBackend,
});

await relay.waitUntilReady();

waitUntilReady() resolves once the node is registered and reading its inbox. You don't have to await it: subscribe and publish wait for it.

On Redis, the relay requires Redis 6.2 or newer (or a compatible server). On PostgreSQL, it requires the 0004_relay migration, for its tables and functions. Run your migrations as usual (see the PostgreSQL backend guide) after upgrading BullMQ, or pass migrate: true as above.

If you supply an existing pg.Pool, set max >= 2: the relay holds one connection for LISTEN and needs another for commands. Allow one listener connection per relay sharing that pool, plus capacity for queries. A pool configuration object can use max: 1, since its listener connects separately.

Subscribing

const subscription = await relay.subscribe('orders.*.status', message => {
  console.log(message.topic); // 'orders.42.status'
  console.log(message.data); // { status: 'shipped' }
});

The first argument is a topic or a pattern: * matches one segment and > matches one or more trailing segments. See Topics and patterns.

The listener receives a RelayMessage:

Field	Description
topic	The topic the message was published on.
data	The published value.
mid	Message id, unique and increasing within the namespace.
ts	Publish timestamp (ms since epoch).
retained	true when it is the topic's retained message.

To stop receiving messages:

await subscription.unsubscribe();

Publishing

const result = await relay.publish('orders.42.status', { status: 'shipped' });
// { mid: '1873', nodes: 3, endpoints: 5 }

data can be any JSON-serializable value. The result tells you how many nodes and subscriptions ("endpoints") the message was delivered to; 0 means nobody was listening.

You can publish from any process, including one that never subscribes.

Closing

Close the relay when the process shuts down, for example next to your workers:

process.on('SIGTERM', async () => {
  await worker.close();
  await relay.close();
});

close() unregisters the node: its subscriptions stop receiving at once and no more messages are written to its inbox. Redis prunes subscription indexes in bounded batches; subsequent sweeps finish any remaining cleanup. If a process exits without closing, other nodes remove it once its lease expires (30 seconds by default); on Redis, if it was the namespace's last node, the namespace expires by itself. See Nodes and failures.

Namespaces

Relays only exchange messages with relays in the same namespace ('default' unless you set one). Use namespaces to isolate environments or applications that share a datastore:

const relay = new Relay({ connection, namespace: 'billing' });

On Redis, the prefix option (default bull) also applies, as for queues.

Errors

subscribe and publish reject with a RelayError when the datastore refuses the operation (for example, an invalid topic), with one of the codes listed in the reference.

Problems that happen in the background, such as a listener throwing or the connection failing while reading the inbox, are emitted as 'error' events. Listen to them, as for any BullMQ class:

relay.on('error', err => logger.error(err, 'relay error'));