Queue consumers (processors) in NestJS - Step-by-Step Execution

Concept Flow - Queue consumers (processors)

Job added to Queue

↓

Queue stores job

↓

Consumer listens to Queue

↓

Consumer receives job

↓

Process job logic

↓

Job completed or failed

↓

Consumer waits for next job

Jobs are added to a queue, consumers listen and pick jobs one by one, process them, then wait for more.

Execution Sample

NestJS

import { Processor, Process } from '@nestjs/bull';
import { Job } from 'bull';

@Processor('email')
export class EmailProcessor {
  @Process('send')
  async handleSendEmail(job: Job) {
    console.log(`Sending email to ${job.data.email}`);
  }
}

This code defines a consumer that listens to the 'email' queue and processes 'send' jobs by logging the email address.

Execution Table

Step	Event	Queue State	Consumer Action	Output
1	Job added to 'email' queue with data {email: '[e-mail]'}	Queue: [{send, {email: '[e-mail]'}}]	Consumer idle, waiting	No output
2	Consumer detects job in queue	Queue: [{send, {email: '[e-mail]'}}]	Consumer picks job	No output
3	Consumer processes job	Queue: []	Runs handleSendEmail with job data	Logs: Sending email to [e-mail]
4	Job completed	Queue: []	Consumer waits for next job	No output
5	Job added to 'email' queue with data {email: '[e-mail]'}	Queue: [{send, {email: '[e-mail]'}}]	Consumer idle, waiting	No output
6	Consumer detects job in queue	Queue: [{send, {email: '[e-mail]'}}]	Consumer picks job	No output
7	Consumer processes job	Queue: []	Runs handleSendEmail with job data	Logs: Sending email to [e-mail]
8	Job completed	Queue: []	Consumer waits for next job	No output
9	No more jobs	Queue: []	Consumer idle	No output

💡 No more jobs in queue, consumer waits idle

Variable Tracker

Variable	Start	After Step 1	After Step 3	After Step 5	After Step 7	Final
Queue	[]	[{send, {email: '[e-mail]'}}]	[]	[{send, {email: '[e-mail]'}}]	[]	[]
Consumer State	Idle	Idle	Processing job	Idle	Processing job	Idle
Current Job Data	null	{email: '[e-mail]'}	null	{email: '[e-mail]'}	null	null

Key Moments - 3 Insights

Why does the consumer wait idle after processing a job?

Because the consumer listens continuously but only acts when a job is in the queue, as shown in steps 4 and 9 where the queue is empty and consumer state is idle.

How does the consumer know which job to process next?

The consumer picks the next job from the queue in order, as seen in steps 2 and 6 where it detects and picks the first job in the queue.

What happens if multiple jobs are added quickly?

Jobs queue up and the consumer processes them one by one in order, as the queue state shows jobs added and removed sequentially in the execution table.

Visual Quiz - 3 Questions

Look at the execution table, what is the consumer state after step 3?

A Processing job

B Idle

C Waiting for job

D Disconnected

💡 Hint

Check the 'Consumer State' variable after step 3 in variable_tracker.

At which step does the queue become empty after processing a job?

A Step 2

B Step 3

C Step 5

D Step 7

💡 Hint

Look at the 'Queue' variable state in variable_tracker after each step.

If a new job is added at step 9, what will the consumer do next?

A Ignore the job

B Crash due to no handler

C Process the job immediately

D Wait until manually restarted

💡 Hint

Refer to the flow where consumer detects and processes jobs as soon as they appear in the queue.

Concept Snapshot

Queue consumers listen to a queue for jobs.
When a job arrives, the consumer picks it up.
It runs the processing function with job data.
After processing, it waits for the next job.
Jobs are processed one at a time in order.
This keeps work organized and scalable.

Full Transcript

In NestJS, queue consumers called processors listen to queues for jobs. When a job is added, the consumer detects it and picks it from the queue. It runs the processing function with the job's data, for example sending an email. After finishing, the consumer waits idle until the next job arrives. This process repeats continuously. The queue holds jobs in order, and the consumer processes them one by one. This helps handle background tasks efficiently and reliably.

Practice

1. What is the main purpose of a queue consumer (processor) in a NestJS application?

easy

A. To send HTTP requests to external APIs

B. To render UI components on the client side

C. To listen to a queue and process jobs asynchronously in the background

D. To manage database migrations synchronously

Solution

1. Step 1: Understand the role of queue consumers

   Queue consumers listen to queues and handle jobs asynchronously, allowing background processing.

2. Step 2: Compare with other options

   Sending HTTP requests, rendering UI, and managing migrations are unrelated to queue consumers.

3. Final Answer:

   To listen to a queue and process jobs asynchronously in the background -> Option C

4. Quick Check:

   Queue consumer = background job handler [OK]

Hint: Queue consumers handle background jobs, not UI or HTTP [OK]

Common Mistakes:

• Confusing queue consumers with HTTP controllers
• Thinking queue consumers run synchronously
• Assuming queue consumers manage UI rendering

2. Which decorator is used to mark a class as a queue consumer (processor) in NestJS?

easy

A. @Processor

B. @Queue

C. @Process

D. @Consumer

Solution

1. Step 1: Identify the decorator for queue consumers

   The @Processor decorator marks a class as a queue consumer in NestJS.

2. Step 2: Differentiate from other decorators

   @Process marks methods, @Queue and @Consumer are not valid NestJS decorators for this purpose.

3. Final Answer:

   @Processor -> Option A

4. Quick Check:

   Class decorator for consumers = @Processor [OK]

Hint: Class decorator for consumers is @Processor, method decorator is @Process [OK]

Common Mistakes:

• Using @Process on the class instead of methods
• Confusing @Processor with @Queue
• Assuming @Consumer is a valid decorator

3. Given the following NestJS queue consumer code, what will be logged when a job with data { name: 'Alice' } is processed?

import { Processor, Process } from '@nestjs/bull';
import { Job } from 'bull';

@Processor('email')
export class EmailProcessor {
  @Process('send')
  async handleSend(job: Job) {
    console.log(`Sending email to ${job.data.name}`);
  }
}

medium

A. No output because handleSend is not called

B. Sending email to Alice

C. Error: job.data is not defined

D. Sending email to undefined

Solution

1. Step 1: Understand job data usage

   The job object contains data with a name property 'Alice'. The method logs this value.

2. Step 2: Confirm method is called on 'send' jobs

   The @Process('send') decorator ensures handleSend runs for jobs named 'send' in the 'email' queue.

3. Final Answer:

   Sending email to Alice -> Option B

4. Quick Check:

   job.data.name = 'Alice' logged correctly [OK]

Hint: Job data is accessed via job.data inside @Process methods [OK]

Common Mistakes:

• Assuming job.data is undefined
• Thinking method won't run without explicit call
• Confusing queue name with job name

4. Identify the error in this NestJS queue consumer code snippet:

import { Processor, Process } from '@nestjs/bull';
import { Job } from 'bull';

@Processor('notifications')
export class NotificationProcessor {
  @Process()
  async handleNotification(job: Job) {
    console.log('Notification:', job.data.message);
  }
}

medium

A. Job parameter type should be any, not Job

B. The class should not have @Processor decorator

C. Console.log syntax is incorrect

D. @Process decorator is missing the job name argument

Solution

1. Step 1: Check @Process decorator usage

   @Process requires a job name string to specify which jobs to handle; it's missing here.

2. Step 2: Validate other parts

   @Processor on class is correct, Job type is valid, console.log syntax is correct.

3. Final Answer:

   @Process decorator is missing the job name argument -> Option D

4. Quick Check:

   @Process needs job name argument [OK]

Hint: @Process must specify job name string inside parentheses [OK]

Common Mistakes:

• Omitting job name in @Process decorator
• Removing @Processor from class
• Changing Job type unnecessarily

5. You want to create a NestJS queue consumer that processes two different job types: 'email' and 'sms' in the same queue 'notifications'. Which is the correct way to define the processor class and methods?

hard

A. Use @Processor('notifications') on class, and two methods with @Process('email') and @Process('sms') decorators

B. Use two separate classes each with @Processor('notifications') and one @Process method

C. Use @Processor('email') and @Processor('sms') on the same class with one @Process method

D. Use @Process('notifications') on methods without @Processor on class

Solution

1. Step 1: Understand queue and job type separation

   The queue is 'notifications'. The job types are 'email' and 'sms'. One class can handle multiple job types with separate @Process decorators.

2. Step 2: Validate correct decorator usage

   @Processor('notifications') marks the queue. Methods use @Process('email') and @Process('sms') to handle each job type.

3. Final Answer:

   Use @Processor('notifications') on class, and two methods with @Process('email') and @Process('sms') decorators -> Option A

4. Quick Check:

   One class, multiple @Process methods for job types [OK]

Hint: One @Processor class can have many @Process methods for different jobs [OK]

Common Mistakes:

• Creating multiple classes for same queue unnecessarily
• Using multiple @Processor decorators on one class
• Omitting @Processor on class and only using @Process