API

QueuePlugin

Binds the QueueManager and mounts the strategies it is given.

app.addPlugin(QueuePlugin, {
  autoStart: true,
  concurrency: 1,
  onUnhandled: 'ignore',
  maxEvents: 50,
  strategies: [defineQueueStrategy({ strategy: MemoryStrategy })],
});

QueueManager

mount()

Mounts a strategy under a name. The strategy is resolved through the container and connects when the manager starts, or on first use when it is mounted after that.

await queueManager.mount({ strategy: MemoryStrategy });
await queueManager.mount({ name: 'events', strategy: KafkaStrategy, initOptions: { client, groupId } });

initOptions is required when the strategy declares initialization options, optional when they include undefined, and rejected when the strategy takes none. The strategy is inferred from strategy, so passing the options of a different strategy is a compile error.

defineQueueStrategy()

Type checks one mount and erases its options type, so a list of mounts of different strategies stays checked.

import { defineQueueStrategy } from '@vercube/queue';

app.addPlugin(QueuePlugin, {
  strategies: [
    defineQueueStrategy({ strategy: RabbitMQStrategy, initOptions: { url: 'amqp://localhost' } }),
    defineQueueStrategy({ name: 'events', strategy: KafkaStrategy, initOptions: { client, groupId } }),
  ],
});

add()

Adds a single job to a queue.

const ref = await queueManager.add({
  queue: 'emails',
  job: 'welcome',
  payload: { userId: '1' },
  options: { attempts: 3, backoff: { type: 'exponential', delay: 1000 }, delay: 5000 },
});

ref.id; // job id, as assigned by the transport

Job options:

addMany()

Adds many jobs of the same kind, using the transport's batch API when it has one.

await queueManager.addMany({
  queue: 'emails',
  job: 'digest',
  payloads: users.map((user) => ({ userId: user.id })),
});

registerConsumer()

Registers a job handler. This is what @Job() calls.

queueManager.registerConsumer({
  strategy: 'default',
  queue: 'emails',
  job: 'welcome',
  handler: async (payload, context) => mailer.send(payload),
  options: { attempts: 3, timeout: 10_000, schema: WelcomeSchema },
  concurrency: 5,
  source: 'EmailConsumer.welcome',
});

Throws when a handler for the same job of the same queue already exists. Use unregisterConsumer({ strategy, queue, job }) to remove one, which also stops the consumer once the queue has no handler left.

Registering under the job name * makes the handler the fallback of its queue, which is what @AnyJob() does. The exported WILDCARD_JOB constant holds that name, so nothing has to be spelled out by hand:

import { WILDCARD_JOB } from '@vercube/queue';

queueManager.registerConsumer({ strategy: 'default', queue: 'events', job: WILDCARD_JOB, handler, options: {}, source: 'ingest' });

registerHook()

Registers a lifecycle hook. This is what @OnJobCompleted() and @OnJobFailed() call.

queueManager.registerHook('failed', {
  strategy: 'default',
  queue: 'emails',
  job: 'welcome',
  hook: async (error, context) => alerts.report(error, context.id),
  source: 'EmailConsumer.failed',
});

unregisterHook(event, registration) removes it again, given the very registration that was registered.

start(), stop() and close()

await queueManager.start(); // connect every strategy and consume every queue that has a handler
await queueManager.stop();  // stop consumers, keep connections open so publishing still works
await queueManager.close(); // stop consumers and close every connection

All three are safe to call more than once. start() runs automatically after the container is initialized unless autoStart is false, and close() runs when the container is torn down.

drain()

Waits for the work the manager scheduled in the background: consumers starting or stopping, and retries waiting for their backoff. Useful in tests and before a deliberate shutdown.

await queueManager.drain();

stats()

Reads the counters a transport keeps for a queue. Transports that keep none return an empty object.

const stats = await queueManager.stats({ queue: 'emails' });

stats.waiting;   // jobs waiting to be processed
stats.active;    // jobs being processed
stats.completed; // completed jobs, on transports that store history
stats.failed;    // jobs that exhausted their attempts
stats.delayed;   // jobs waiting for their delay

subscribe()

Follows every job the manager finishes, as it finishes it, instead of polling inspect(). This is how the devtools panel stays live.

const stop = queueManager.subscribe((event) => {
  if (event.status === 'failed') {
    metrics.increment(`queue.${event.queue}.${event.job}.failed`);
  }
});

stop(); // removes the listener again

The listener gets the same event the inspection buffer keeps, so it carries the outcome, the attempt, the duration and, for a failure, the error with its stack. It is called even when maxEvents is 0, so a live view keeps working without any buffer at all. A listener that throws is logged and kept, and never changes the outcome of the job.

peek()

Shows what a queue is holding, without consuming any of it.

const messages = await queueManager.peek({ queue: 'emails', limit: 25, states: ['waiting', 'delayed', 'failed'] });

messages[0].state;      // 'waiting', 'active', 'delayed' or 'failed'
messages[0].job;        // name of the job
messages[0].payload;    // preview, with credential-looking fields withheld
messages[0].availableAt; // when a delayed message becomes available
messages[0].error;      // why it failed, for a message in a failed set

Only transports that can be read without side effects support this, which they report as the peek capability: memory and BullMQ do, RabbitMQ and Kafka do not and return nothing. On BullMQ the failed set is the interesting one, because Redis keeps both the data and the stack trace of every job that ran out of attempts.

inspect()

Describes what the module holds: mounted strategies, registered handlers, per-queue counters and the last processed jobs. This is what the devtools panel reads.

const snapshot = queueManager.inspect();

snapshot.started;    // whether consumers are running
snapshot.strategies; // one entry per mount, with its capabilities and status
snapshot.consumers;  // one entry per registered handler
snapshot.metrics;    // published, processed, failed, retried, unhandled and active, per queue
snapshot.events;     // recently processed jobs, newest first

configure()

Sets manager-wide settings, merging into the existing ones. Per-job and per-handler options always win.

queueManager.configure({ concurrency: 4, onUnhandled: 'fail', maxEvents: 200 });

getStrategy()

Returns a mounted strategy, for the rare case a transport-specific API is needed.

const bullmq = queueManager.getStrategy() as BullMQStrategy;

Decorators

See Consumers for the full description of each.

JobContext

The second argument of every handler and hook.

QueueError

Every failure the module raises itself is a QueueError. What a handler throws is passed through untouched, so a handler that throws a plain Error fails the attempt with that error.

try {
  await queueManager.add({ queue: 'emails', job: 'welcome', payload });
} catch (error) {
  if (error instanceof QueueError) {
    error.operation; // 'add', 'publish', 'consume', 'validate', 'timeout', ...
    error.retryable; // false when running the job again cannot help
    error.cause;     // the transport error underneath
    error.metadata;  // queue, job and other non-sensitive context
  }
}

retryable is what the manager reads when deciding whether to schedule another attempt: a validation failure or a missing handler is never retried, while a broker hiccup is.

Job headers

Three headers travel with every job so any transport can route and retry it. They are visible in context.headers and can be read by other consumers of the same broker.

A retry is a fresh publish, so it is these headers that carry the key and the priority of the original across it.

A message that carries no job name, because it was produced outside this module and its transport's own field is empty too, is dispatched as the job unknown, which an @AnyJob() handler picks up.

x-attempts is read off the wire on transports that do not retry natively, so the budget it asks for is capped at 50 and a backoff is never held longer than a day.