QueueEvents

Listen to global events emitted by a queue

Overview

The QueueEvents class is used for listening to global events emitted by a queue. This class requires a dedicated Redis connection.

Constructor

new QueueEvents(
  name: string,
  opts?: QueueEventsOptions
)

name · string · required

The name of the queue to listen to

opts · QueueEventsOptions

Configuration options for queue events

Example

import { QueueEvents } from 'bullmq';

const queueEvents = new QueueEvents('myQueue', {
  connection: {
    host: 'localhost',
    port: 6379,
  },
});

queueEvents.on('completed', ({ jobId, returnvalue }) => {
  console.log(`Job ${jobId} completed with result:`, returnvalue);
});

queueEvents.on('failed', ({ jobId, failedReason }) => {
  console.log(`Job ${jobId} failed:`, failedReason);
});

Methods

run

Manually starts the event consumption loop.

run(): Promise<void>

You only need to call this if you set autorun: false in the constructor options.

close

Stops consuming events and closes the Redis connection.

close(): Promise<void>

Events

The QueueEvents class extends EventEmitter and emits the following events:

added

Emitted when a job is created and added to the queue.

queueEvents.on('added', ({ jobId, name }, id) => {
  console.log(`Job ${jobId} (${name}) was added`);
});

args.jobId · string

The unique identifier of the job

args.name · string

The name of the job

id · string

The event stream ID

active

Emitted when a job enters the active state.

queueEvents.on('active', ({ jobId, prev }, id) => {
  console.log(`Job ${jobId} is now active`);
});

args.jobId · string

The unique identifier of the job

args.prev · string

The previous state of the job

completed

Emitted when a job has successfully completed.

queueEvents.on('completed', ({ jobId, returnvalue, prev }, id) => {
  console.log(`Job ${jobId} completed:`, returnvalue);
});

args.jobId · string

The unique identifier of the job

args.returnvalue · any

The return value of the job (JSON parsed)

args.prev · string

The previous state of the job

failed

Emitted when a job has failed.

queueEvents.on('failed', ({ jobId, failedReason, prev }, id) => {
  console.log(`Job ${jobId} failed:`, failedReason);
});

args.jobId · string

The unique identifier of the job

args.failedReason · string

The reason or message describing why the job failed

args.prev · string

The previous state of the job

progress

Emitted when a job updates its progress.

queueEvents.on('progress', ({ jobId, data }, id) => {
  console.log(`Job ${jobId} progress:`, data);
});

args.jobId · string

The unique identifier of the job

args.data · number | object

The progress data

delayed

Emitted when a job is scheduled with a delay.

queueEvents.on('delayed', ({ jobId, delay }, id) => {
  console.log(`Job ${jobId} delayed by ${delay}ms`);
});

args.jobId · string

The unique identifier of the job

args.delay · number

The delay duration in milliseconds

waiting

Emitted when a job enters the waiting state.

queueEvents.on('waiting', ({ jobId, prev }, id) => {
  console.log(`Job ${jobId} is waiting`);
});

waiting-children

Emitted when a job is waiting for its children to complete.

queueEvents.on('waiting-children', ({ jobId }, id) => {
  console.log(`Job ${jobId} waiting for children`);
});

removed

Emitted when a job is removed from the queue.

queueEvents.on('removed', ({ jobId, prev }, id) => {
  console.log(`Job ${jobId} removed`);
});

drained

Emitted when the queue has drained its waiting list.

queueEvents.on('drained', (id) => {
  console.log('Queue is drained');
});

paused

Emitted when the queue is paused.

queueEvents.on('paused', ({}, id) => {
  console.log('Queue paused');
});

resumed

Emitted when the queue is resumed.

queueEvents.on('resumed', ({}, id) => {
  console.log('Queue resumed');
});

stalled

Emitted when a job has stalled.

queueEvents.on('stalled', ({ jobId }, id) => {
  console.log(`Job ${jobId} stalled`);
});

retries-exhausted

Emitted when a job has exhausted its retry attempts.

queueEvents.on('retries-exhausted', ({ jobId, attemptsMade }, id) => {
  console.log(`Job ${jobId} exhausted retries`);
});

duplicated

Emitted when a job is not created because a job with the same ID already exists.

queueEvents.on('duplicated', ({ jobId }, id) => {
  console.log(`Job ${jobId} duplicated`);
});

deduplicated

Emitted when a job is not added because a job with the same deduplication ID exists.

queueEvents.on('deduplicated', ({ jobId, deduplicationId, deduplicatedJobId }, id) => {
  console.log(`Job ${jobId} deduplicated`);
});

cleaned

Emitted when jobs are cleaned from the queue.

queueEvents.on('cleaned', ({ count }, id) => {
  console.log(`${count} jobs cleaned`);
});

error

Emitted when an error occurs in the Redis backend.

queueEvents.on('error', (error: Error) => {
  console.error('Queue events error:', error);
});

Job-Specific Events

You can also listen to events for a specific job:

const jobId = 'my-job-id';

queueEvents.on(`completed:${jobId}`, ({ returnvalue }, id) => {
  console.log('My specific job completed:', returnvalue);
});

queueEvents.on(`failed:${jobId}`, ({ failedReason }, id) => {
  console.log('My specific job failed:', failedReason);
});