This class provides an efficient and flexible way to handle batch processing of jobs in a BullMQ environment, leveraging NestJS. It ensures that jobs are processed in batches, which can be crucial for performance optimization in applications requiring bulk operations..

batch-worker-host.ts

import {WorkerHost} from '@nestjs/bullmq';
import {Job} from 'bullmq';

import {GeneralLogger} from '../log/general-logger';

export class BatchWorkerHost extends WorkerHost {
    private jobBatch: Job[];
    private jobBatchProcessPromise: Promise<void>;

    private resolveJobBatchProcessPromise: () => void;
    private rejectJobBatchProcessPromise: (error: Error) => void;

    private readonly logger: GeneralLogger;

    constructor(private readonly batchSize: number, private readonly batchInterval: number) {
        super();

        this.logger = new GeneralLogger(this.constructor.name);

        this.initializeJobBatch();

        this.startBatchProcessing();
    }

    initializeJobBatch(): void {
        this.jobBatch = [];

        this.jobBatchProcessPromise = new Promise<void>((resolve, reject) => {
            this.resolveJobBatchProcessPromise = resolve;
            this.rejectJobBatchProcessPromise = reject;
        });
    }

    async process(job: Job): Promise<void> {
        this.jobBatch.push(job);

        const jobBatchProcessPromise = this.jobBatchProcessPromise;

        if (this.jobBatch.length >= this.batchSize) {
            this.processBatchWrapper();
        }

        await jobBatchProcessPromise;
    }

    private startBatchProcessing(): void {
        setInterval(() => {
            if (this.jobBatch.length > 0) {
                this.processBatchWrapper();
            }
        }, this.batchInterval);
    }

    private processBatchWrapper(): void {
        const jobBatch = this.jobBatch;
        const resolveJobBatchProcessPromise = this.resolveJobBatchProcessPromise;
        const rejectJobBatchProcessPromise = this.rejectJobBatchProcessPromise;

        this.initializeJobBatch();

        this.processBatch(jobBatch)
            .then(() => {
                resolveJobBatchProcessPromise();
            })
            .catch(error => {
                this.logger.error('Error processing batch', error.stack);

                rejectJobBatchProcessPromise(error);
            });
    }

    protected async processBatch(jobs: Job[]): Promise<void> {
        throw new Error('You must implement the processBatch method in your subclass.');
    }
}

Usage

To utilize this class, create a subclass and implement the processBatch method to define how each batch of jobs should be processed.

import { BatchWorkerHost } from './batch-worker-host';
import { Job } from 'bullmq';

class MyCustomBatchWorker extends BatchWorkerHost {
    constructor() {
        super(200, 5000); // Batch size of 200 and interval of 5000ms
    }

    protected async processBatch(jobs: Job[]): Promise<void> {
        // Custom batch processing logic
        for (const job of jobs) {
            // Process each job
        }
    }
}