Consumers

A consumer is a container service that reads one queue. The class decorator says which queue, the method decorators say which jobs.

@Consumer

Marks a class as the consumer of a queue and sets the defaults for every handler in it.

@Consumer({ queue: 'emails', concurrency: 5, attempts: 2, timeout: 10_000 })
export class EmailConsumer {}

The queue starts being consumed when the class is bound in the container:

container.bind(EmailConsumer);

Nothing happens without that binding, exactly like a controller that is never bound. A handler declared on a subclass inherits the queue of its base class, so shared consumer bases work as expected.

@Job

Marks a method as the handler of one job name. The method receives the payload and the job context.

@Consumer({ queue: 'emails' })
export class EmailConsumer {
  @Job('welcome')
  public async welcome(payload: { userId: string }, context: QueueTypes.JobContext): Promise<void> {
    context.logger?.info(`attempt ${context.attempt} of ${context.attempts}`);

    await this.gMailer.sendWelcome(payload.userId);
  }
}

Options given here override the ones the class declared:

Returning marks the job done, throwing marks the attempt failed. There is nothing to acknowledge by hand.

Where the job name comes from

A job name is not a broker concept. It is a label the module writes into the message metadata, so that one queue can carry several kinds of work, each with its own handler, retries and schema. Every transport keeps it where it naturally belongs:

So a job published as add({ queue: 'emails', job: 'welcome', payload: { userId } }) arrives on RabbitMQ as an ordinary AMQP message whose body is the payload and whose type is welcome:

queue: emails
properties:
  type:      'welcome'
  messageId: '0f2c…'
  headers:   { x-job: 'welcome', x-attempt: '1' }
body: {"userId":"u-42"}

If a queue only ever carries one kind of work, the name is close to cosmetic: declare a single @Job('process') and be done. Modelling it the other way round, one queue per job, works just as well.

A message carrying none of that metadata arrives as the job unknown, which is what @AnyJob() is for. A foreign producer that does set the field its transport uses, an AMQP type or a Kafka x-job header, names its jobs just as this module does, and a @Job() handler picks them up by name.

@AnyJob

Handles every job of the queue that no @Job() claims. A named handler always wins, so this is a fallback rather than a replacement, and there can be one per queue.

// a queue another application fills, whose messages carry no job name
@Consumer({ queue: 'legacy-events' })
export class LegacyConsumer {
  @AnyJob({ attempts: 3 })
  public async handle(payload: unknown, context: QueueTypes.JobContext): Promise<void> {
    // context.job is whatever arrived, or 'unknown' when nothing did
    await this.gEvents.ingest(payload, context.job);
  }
}

It takes the same options as @Job(), and they apply to every job it catches:

@Consumer({ queue: 'emails' })
export class EmailConsumer {
  @Job('welcome')
  public async welcome(payload: Welcome): Promise<void> {}

  @AnyJob()
  public async rest(payload: unknown, context: QueueTypes.JobContext): Promise<void> {
    context.logger?.warn(`unrecognised job ${context.job}`);
  }
}

The real job name is on the context, is what the retry republishes under, and is what the counters and the devtools panel report. The fallback never changes a job's identity, it only decides who runs it.

The job context

The second argument describes the attempt being processed.

@Job('report')
public async report(payload: Report, context: QueueTypes.JobContext<Report>): Promise<void> {
  // where it came from
  context.queue;     // 'reports'
  context.job;       // 'report'
  context.strategy;  // name of the mounted strategy
  context.id;        // job id, as assigned by the transport
  context.headers;   // transport headers received with the job

  // which attempt this is
  context.attempt;   // 1 on the first run
  context.attempts;  // how many attempts this job may take

  // reporting back
  context.logger?.info('halfway');
  await context.updateProgress(50);

  // the transport-native job, when you really need it
  context.raw;
}

updateProgress() is a no-op on transports that do not track progress, so calling it is always safe. context.logger is the application logger with the queue, job, id and attempt already attached, and is null when no logger is bound.

Validating payloads

A job payload comes from outside the process, so it deserves the same treatment as a request body. Pass any Standard Schema and the payload is validated before the handler runs.

import { z } from 'zod';

const WelcomeSchema = z.object({ userId: z.string().uuid() });

@Consumer({ queue: 'emails' })
export class EmailConsumer {
  @Job('welcome', { schema: WelcomeSchema })
  public async welcome(payload: z.infer<typeof WelcomeSchema>): Promise<void> {}
}

The handler receives the parsed value, so defaults and coercions the schema applies are visible inside it. A payload that does not match fails the job without retrying it.

Lifecycle hooks

@OnJobCompleted() and @OnJobFailed() observe the queue of their consumer class. A hook never changes the outcome of a job: if a hook throws, the throw is logged and forgotten.

@Consumer({ queue: 'emails' })
export class EmailConsumer {
  @Job('welcome', { attempts: 3 })
  public async welcome(payload: { userId: string }): Promise<void> {}

  @OnJobCompleted()
  public async sent(context: QueueTypes.JobContext): Promise<void> {
    this.gMetrics.increment(`emails.${context.job}.sent`);
  }

  @OnJobFailed()
  public async failed(error: Error, context: QueueTypes.JobContext): Promise<void> {
    // called once per failed attempt, so tell a retry from a final failure
    if (context.attempt === context.attempts) {
      await this.gAlerts.report(error, context.id);
    }
  }
}

Both accept a job filter when a hook should only watch one kind of job:

@OnJobFailed({ job: 'digest' })
public async digestFailed(error: Error): Promise<void> {}

Concurrency

concurrency is per queue and per process. The highest value requested by any handler of a queue wins, and the manager's concurrency setting is the floor for every queue.

// up to 10 jobs of 'emails' at a time in this process
@Consumer({ queue: 'emails', concurrency: 10 })
export class EmailConsumer {}

On BullMQ it becomes the worker concurrency, on RabbitMQ the channel prefetch, on Kafka the number of partitions consumed concurrently, and in memory the number of handlers running at once.

Registering handlers by hand

The decorators are a thin layer over QueueManager.registerConsumer(). Consumers built at runtime can call it directly:

queueManager.registerConsumer({
  strategy: 'default',
  queue: 'emails',
  job: 'welcome',
  handler: async (payload, context) => this.send(payload),
  options: { attempts: 3 },
  concurrency: 2,
  source: 'runtime',
});

Registering a second handler for the same job of the same queue is an error, so a copy-pasted job name fails loudly instead of silently shadowing the first handler.