BullMQ Module

The @grupodiariodaregiao/bunstone BullMQ module provides a way to process background jobs using BullMQ.

Installation

First, ensure you have the required dependencies:

bun add bullmq ioredis

Setup

To use BullMQ, you need to register the BullMqModule in your AppModule.

import { BullMqModule, Module } from "@grupodiariodaregiao/bunstone";

@Module({
  imports: [
    BullMqModule.register({
      host: 'localhost',
      port: 6379,
    }),
  ],
})
export class AppModule {}

Creating a Processor

A processor is a class decorated with @Processor(). It contains methods decorated with @Process() that handle jobs from a specific queue.

import { Process, Processor } from "@grupodiariodaregiao/bunstone";
import { Job } from "bullmq";

@Processor('email-queue')
export class EmailProcessor {
  @Process('send-welcome')
  async handleSendWelcome(job: Job) {
    console.log('Sending welcome email to:', job.data.email);
    // Simulate work
    await new Promise(resolve => setTimeout(resolve, 1000));
    return { sent: true };
  }

  @Process()
  async handleDefault(job: Job) {
    console.log('Handling unknown job type:', job.name);
  }
}

Don't forget to add your processor to the providers of a module.

Producing Jobs

You can inject the QueueService into your controllers or services to add jobs to a queue.

import {
  Controller,
  Get,
  Query,
  QueueService,
} from "@grupodiariodaregiao/bunstone";

@Controller('/email')
export class EmailController {
  constructor(private readonly queueService: QueueService) {}

  @Get('/send')
  async sendEmail(@Query('email') email: string) {
    await this.queueService.add('email-queue', 'send-welcome', { email });
    return { status: 'Job added to queue' };
  }
}

Full Example

Explore a complete producer + processor flow:

import {
	AppStartup,
	BullMqModule,
	Controller,
	Get,
	Module,
	Process,
	Processor,
	QueueService,
} from "../../index";
import { Job } from "bullmq";

// 1. Define a Job Processor
@Processor({
	queueName: "mail-queue",
	concurrency: 2,
})
export class MailProcessor {
	@Process("welcome-email")
	async handleWelcomeEmail(job: Job) {
		console.log(`[Worker] Processing welcome email for: ${job.data.email}`);
		// Simulate some work
		await new Promise((resolve) => setTimeout(resolve, 1000));
		console.log(`[Worker] Welcome email sent to ${job.data.email}`);
		return { success: true, recipient: job.data.email };
	}

	@Process()
	async handleGenericJob(job: Job) {
		console.log(`[Worker] Processing generic job: ${job.name}`);
	}
}

// 2. Define a Controller to produce jobs
@Controller("/jobs")
export class JobsController {
	constructor(private readonly queueService: QueueService) {}

	@Get("/add")
	async addJob() {
		const email = `user-${Math.floor(Math.random() * 1000)}@example.com`;
		console.log(`[Controller] Adding welcome-email job for ${email}`);

		await this.queueService.add("mail-queue", "welcome-email", { email });

		return {
			message: "Job added to queue",
			email,
		};
	}
}

// 3. Setup the Application Module
@Module({
	imports: [
		BullMqModule.register({
			host: process.env.REDIS_HOST || "localhost",
			port: Number(process.env.REDIS_PORT) || 6379,
		}),
	],
	controllers: [JobsController],
	providers: [MailProcessor],
})
class AppModule {}

// 4. Start the app
console.log("Starting BullMQ example app...");
console.log("Make sure you have a Redis instance running at localhost:6379");
const app = await AppStartup.create(AppModule);
app.listen(3000);

See it on GitHub

Processor Options

The @Processor decorator accepts an options object:

@Processor({
  queueName: 'heavy-tasks',
  concurrency: 5,
})
export class HeavyTaskProcessor {
  // ...
}