NestJS Queue Events

Listen to queue events in NestJS using QueueEventsListener

Queue events allow you to monitor what’s happening in your queues without the overhead of running a full worker. Use the QueueEventsListener to listen to events like job completion, failure, and progress updates.

Basic Setup

1. Create a QueueEventsListener class

   Use the @QueueEventsListener decorator to create an events listener:

   import {
     QueueEventsListener,
     QueueEventsHost,
     OnQueueEvent,
   } from '@nestjs/bullmq';
   
   @QueueEventsListener('audio')
   export class AudioQueueEvents extends QueueEventsHost {
     @OnQueueEvent('completed')
     onCompleted({
       jobId,
       returnvalue,
     }: {
       jobId: string;
       returnvalue: string;
       prev?: string;
     }) {
       console.log(`Job ${jobId} completed with result:`, returnvalue);
     }
   }
   
2. Register as a provider

   Add the events listener to your module’s providers:

   import { Module } from '@nestjs/common';
   import { BullModule } from '@nestjs/bullmq';
   import { AudioQueueEvents } from './audio.queue-events';
   
   @Module({
     imports: [
       BullModule.registerQueue({
         name: 'audio',
         connection: {
           host: 'localhost',
           port: 6379,
         },
       }),
     ],
     providers: [AudioQueueEvents],
   })
   export class AudioModule {}
   

Available Events

Listen to various queue events using the @OnQueueEvent decorator:

completed

Triggered when a job completes successfully:

@OnQueueEvent('completed')
onCompleted({
  jobId,
  returnvalue,
  prev,
}: {
  jobId: string;
  returnvalue: string;
  prev?: string;
}) {
  console.log(`Job ${jobId} completed`);
  console.log('Result:', returnvalue);
}

failed

Triggered when a job fails:

@OnQueueEvent('failed')
onFailed({
  jobId,
  failedReason,
  prev,
}: {
  jobId: string;
  failedReason: string;
  prev?: string;
}) {
  console.error(`Job ${jobId} failed:`, failedReason);
}

progress

Triggered when a job reports progress:

@OnQueueEvent('progress')
onProgress({
  jobId,
  data,
}: {
  jobId: string;
  data: number | object;
}) {
  console.log(`Job ${jobId} progress:`, data);
}

active

Triggered when a job becomes active:

@OnQueueEvent('active')
onActive({
  jobId,
  prev,
}: {
  jobId: string;
  prev?: string;
}) {
  console.log(`Job ${jobId} started processing`);
}

Complete Example

audio.queue-events.ts

import {
  QueueEventsListener,
  QueueEventsHost,
  OnQueueEvent,
} from '@nestjs/bullmq';
import { Injectable, Logger } from '@nestjs/common';

@QueueEventsListener('audio')
export class AudioQueueEvents extends QueueEventsHost {
  private readonly logger = new Logger(AudioQueueEvents.name);
  
  @OnQueueEvent('completed')
  onCompleted({
    jobId,
    returnvalue,
  }: {
    jobId: string;
    returnvalue: string;
    prev?: string;
  }) {
    this.logger.log(`Audio job ${jobId} completed successfully`);
    this.logger.debug('Result:', returnvalue);
  }
  
  @OnQueueEvent('failed')
  onFailed({
    jobId,
    failedReason,
  }: {
    jobId: string;
    failedReason: string;
    prev?: string;
  }) {
    this.logger.error(`Audio job ${jobId} failed: ${failedReason}`);
  }
  
  @OnQueueEvent('progress')
  onProgress({
    jobId,
    data,
  }: {
    jobId: string;
    data: number | object;
  }) {
    this.logger.debug(`Audio job ${jobId} progress:`, data);
  }
  
  @OnQueueEvent('active')
  onActive({
    jobId,
  }: {
    jobId: string;
    prev?: string;
  }) {
    this.logger.log(`Audio job ${jobId} started processing`);
  }
  
  @OnQueueEvent('waiting')
  onWaiting({ jobId }: { jobId: string }) {
    this.logger.debug(`Audio job ${jobId} is waiting`);
  }
  
  @OnQueueEvent('delayed')
  onDelayed({
    jobId,
    delay,
  }: {
    jobId: string;
    delay: number;
  }) {
    this.logger.log(`Audio job ${jobId} delayed by ${delay}ms`);
  }
  
  @OnQueueEvent('stalled')
  onStalled({ jobId }: { jobId: string }) {
    this.logger.warn(`Audio job ${jobId} stalled`);
  }
}

Event Types

All available queue events:

Job Lifecycle Events

• waiting - Job is waiting to be processed
• active - Job has started processing
• completed - Job completed successfully
• failed - Job failed
• progress - Job reported progress
• delayed - Job is delayed
• stalled - Job stalled (worker lost lock)

Queue Management Events

• paused - Queue was paused
• resumed - Queue was resumed
• cleaned - Old jobs were cleaned
• drained - Queue was drained

Job Management Events

• removed - Job was removed
• retries-exhausted - Job exhausted all retry attempts

Multiple Queue Listeners

Create separate event listeners for different queues:

// audio.queue-events.ts
@QueueEventsListener('audio')
export class AudioQueueEvents extends QueueEventsHost {
  @OnQueueEvent('completed')
  onCompleted(args: any) {
    // Handle audio queue completion
  }
}

// video.queue-events.ts
@QueueEventsListener('video')
export class VideoQueueEvents extends QueueEventsHost {
  @OnQueueEvent('completed')
  onCompleted(args: any) {
    // Handle video queue completion
  }
}

// app.module.ts
@Module({
  imports: [
    BullModule.registerQueue({ name: 'audio' }),
    BullModule.registerQueue({ name: 'video' }),
  ],
  providers: [AudioQueueEvents, VideoQueueEvents],
})
export class AppModule {}

Integration with Services

Inject services into your event listeners:

import {
  QueueEventsListener,
  QueueEventsHost,
  OnQueueEvent,
} from '@nestjs/bullmq';
import { Injectable } from '@nestjs/common';
import { NotificationService } from './notification.service';

@Injectable()
@QueueEventsListener('orders')
export class OrderQueueEvents extends QueueEventsHost {
  constructor(private notificationService: NotificationService) {
    super();
  }
  
  @OnQueueEvent('completed')
  async onCompleted({
    jobId,
    returnvalue,
  }: {
    jobId: string;
    returnvalue: any;
  }) {
    // Send notification when order is processed
    await this.notificationService.sendOrderConfirmation(
      returnvalue.orderId
    );
  }
  
  @OnQueueEvent('failed')
  async onFailed({
    jobId,
    failedReason,
  }: {
    jobId: string;
    failedReason: string;
  }) {
    // Alert admins about failed orders
    await this.notificationService.alertAdmins(
      `Order job ${jobId} failed: ${failedReason}`
    );
  }
}

Best Practices

Use Logger

Use NestJS Logger for consistent logging:

private readonly logger = new Logger(MyEvents.name);

Keep Handlers Fast

Event handlers should be quick - delegate heavy work to other services

Error Handling

Handle errors in event listeners:

@OnQueueEvent('completed')
async onCompleted(args: any) {
  try {
    await this.doSomething();
  } catch (error) {
    this.logger.error(error);
  }
}

Separate Concerns

Use different event listeners for different responsibilities

Related Resources

NestJS Integration

Learn about NestJS integration basics

NestJS Producers

Add jobs to queues in NestJS

External Documentation

NestJS Queues

Official NestJS queue documentation

Queue Events API

BullMQ Queue Events API reference