Implementing Bull Queue in Typescript

Question

I try to implement Bull queue in Typescript and NestJS, my code:

@Injectable()
export class MailService {
    constructor(
        @InjectQueue('mail')
        private readonly mailQueue: Queue
    ) {}

    async addToQueue(): Promise<void> {
        this.mailQueue.add(() => {
            return this.sendMail(); 
        })
    }
    
    
    async sendMail(): Promise<void> {

        //logic to implement

        this.addToQueue();
    }
}

fast question: Is this implementation sufficient for my job queuing to work?, If not: what i must to do?

score 3 · Accepted Answer · edited Jul 15 '20 at 23:46

I recently wrote a blog post that seems to relate to your use-case:

https://firxworx.com/blog/coding/nodejs/email-module-for-nestjs-with-bull-queue-and-the-nest-mailer/

A few tips:

In your module, be sure to import your BullModule (from @nestjs/bull). For example, you need to configure with your queue name ("mail" in your case) and setup your queue. A common setup would include configuring with the redis hostname and port.
In your service, you need to add jobs to the queue, along with optional payload. In your case, you are trying to add a function. Instead, you should add a job name, e.g. "confirmationEmail", and pass a payload, e.g.user and token. My example would look like this: await this.mailQueue.add('confirmationEmail', { user, token })
You need to implement a processor for your queue. This is a class decorated with the @Processor(QUEUE_NAME) decorator from @nestjs/bull (@Processor('mail') in your case). The processor handles jobs that are added to the queue.
In your processor, you could implement a method e.g. sendConfirmationEmail() that handles the job named "confirmationEmail". You would decorate that method with @Process(JOB_NAME), e.g. @Process('confirmationEmail'). The method can receive your payload. Per my example, the following method signature would provide the user and token: async sendConfirmationEmail(job: Job<{ user: User, token: string }>): Promise<any> (note Job is from the bull package, and that you may wish to type your return vs. using any). Here is where you would actually send out the email.
In your processor class, @nestjs/bull also provides special method decorators including @OnQueueActive(), @OnQueueCompleted(), @OnQueueFailed(). Refer to the docs but you may find these useful for logging or other purposes.

The idea is that your processor handles jobs in the queue when the app is otherwise idle.

Your mail module would presumably have at least a mail.module.ts with configuration, a mail.service.ts that adds jobs to the "mail" queue, and a mail.processor.ts that takes care of completing any jobs added to the "mail" queue.

Further documentation from NestJS is available at:

https://docs.nestjs.com/techniques/queues

Great! Thanks man!, one additional question, how can i handle logic in e.q. @OnQueueActive instead of logging? — , Jul 16 '20 at 08:24
Hey I'm glad it helped. The method decorated with `@OnQueueActive()` will be passed the job as a param (e.g. method signature: `onQueueActive(job: Job)`). Therefore inside the method you have access to `job.id`, `job.name`, `job.data` and can go from there. I don't think you necessarily need to put logic in there though, as you could write a `@Process()` decorated method in most cases to break up your code a bit better. — firxworx, Jul 16 '20 at 15:51
To round out my comment, `@OnQueueCompleted()` decorated methods receive the `job` and `result` as params, and `@OnQueueFailed()` decorated methods receive the `job` and `error` as params. — firxworx, Jul 16 '20 at 15:56
i have one last question, can you show an example of tests of the this bull queue in Jest? — , Jul 23 '20 at 20:56
@Qumber does the IoC work with workers? i'm trying to resolve a dependency but it fail although the service is in the same module with the processor class. — joe_inz, Mar 20 '21 at 11:04

Implementing Bull Queue in Typescript

1 Answers1

Linked