Queue

Contract

@modularityjs/queue defines the abstract QueueService and @Consume decorator:

abstract class QueueService {
  abstract publish<T>(
    topic: string,
    payload: T,
    options?: PublishOptions,
  ): Promise<void>;
  abstract publishBatch<T>(
    messages: Array<{ topic: string; payload: T; options?: PublishOptions }>,
  ): Promise<void>;
  abstract getDeadLetters(topic?: string): Promise<QueueMessage[]>;
  abstract purgeDeadLetters(topic?: string): Promise<number>;
}

Drivers

Memory (@modularityjs/queue-memory)

In-memory queue with immediate dispatch. For development and testing.

import { QueueModule } from '@modularityjs/queue';
import { QueueMemoryModule } from '@modularityjs/queue-memory';

const modules = [QueueModule, QueueMemoryModule];

Redis (@modularityjs/queue-redis)

Redis Streams-backed queue with consumer groups, retry with backoff, dead letter queues, and delayed messages. Requires @modularityjs/redis.

import { QueueModule } from '@modularityjs/queue';
import { QueueRedisModule } from '@modularityjs/queue-redis';
import { RedisModule } from '@modularityjs/redis';

const modules = [
  RedisModule.forRoot({ url: 'redis://localhost:6379' }),
  QueueModule,
  QueueRedisModule,
];

RabbitMQ (@modularityjs/queue-rabbitmq)

AMQP-backed queue (via the encapsulated rabbitmq-client package). Each topic maps to three durable queues: a work queue, a .delay queue, and a .dlq dead-letter queue. Delayed delivery and retry backoff both ride the delay queue: messages are published with a per-message TTL and the queue dead-letters them into the work queue when the TTL expires (TTL+DLX pattern).

import { QueueModule } from '@modularityjs/queue';
import { QueueRabbitmqModule } from '@modularityjs/queue-rabbitmq';

const modules = [
  QueueModule,
  QueueRabbitmqModule.forRoot({ url: 'amqp://localhost:5672' }),
];

Option	Default	Description
url	— (required)	amqp:// or amqps:// broker URL
queueNamePrefix	'modularityjs.queue.'	Prefix for every declared queue name
prefetchCount	10	Per-topic basic.qos prefetch
deadLetterCapacity	10_000	x-max-length cap on each DLQ (drop-head overflow); 0 = uncapped
queueType	'classic'	'classic' or 'quorum'

Retries are per-consumer: a retried message carries the list of consumers that still need it, so siblings that already succeeded on the same topic are not re-run. Exhausted messages land in the topic's DLQ (with a ModularityJsQueueDeadLetter warning); getDeadLetters reads it non-destructively, purgeDeadLetters empties it.

TTL+DLX head-of-line caveat

AMQP expires messages only at the head of a queue — a long-delay message can park shorter-delay ones behind it. Delays are never early but can be late. If you need precise timers at scale, use queue-redis (sorted-set delayed delivery).

Publishing

import { Inject, Injectable } from '@modularityjs/di';
import { QueueService } from '@modularityjs/queue';

@Injectable()
class OrderService {
  constructor(@Inject(QueueService) private readonly queue: QueueService) {}

  async placeOrder(order: Order): Promise<void> {
    await this.queue.publish('order.placed', order);
  }
}

Consuming

Consumers use the @Consume decorator. Register the consumer class in the QueueConsumersPool:

import { Consume, QueueConsumersPool } from '@modularityjs/queue';

@Injectable()
class OrderConsumer {
  @Consume({ topic: 'order.placed', name: 'send-confirmation' })
  async handleOrderPlaced(message: QueueMessage<Order>): Promise<void> {
    await sendConfirmationEmail(message.payload);
  }
}

@Module({
  name: 'order-consumers',
  imports: [QueueModule],
  providers: [OrderConsumer],
  pools: [
    {
      pool: QueueConsumersPool,
      key: 'order-consumer',
      useClass: OrderConsumer,
    },
  ],
})
class OrderConsumersModule {}

Retry Policy

@Consume supports configurable retry:

@Consume({
  topic: 'order.placed',
  name: 'send-confirmation',
  retryPolicy: {
    maxAttempts: 5,
    strategy: 'exponential', // or 'fixed'
    delayMs: 1000,
    maxDelayMs: 30_000,
  },
  concurrency: 3,
})

Messages that exhaust all retries are moved to the dead letter queue.

Configuration

QueueModule:

QueueModule.forRoot({
  defaultRetryPolicy: {
    maxAttempts: 5,
    strategy: 'exponential',
    delayMs: 2000,
    maxDelayMs: 60_000,
  },
  shutdownTimeoutMs: 10_000,
});

Option	Default	Description
defaultRetryPolicy.maxAttempts	3	Maximum delivery attempts before dead-lettering
defaultRetryPolicy.strategy	'exponential'	'fixed' or 'exponential' backoff
defaultRetryPolicy.delayMs	1000	Base delay between retries in ms
defaultRetryPolicy.maxDelayMs	30000	Upper bound for exponential backoff in ms
shutdownTimeoutMs	30000	Grace period for in-flight messages during shutdown

QueueRedisModule:

QueueRedisModule.forRoot({
  keyNamespace: 'queue:',
  consumerGroup: 'default',
  blockTimeoutMs: 5000,
  batchSize: 10,
});

Option	Default	Description
keyNamespace	'queue:'	Prefix for Redis Stream keys
consumerGroup	'default'	Redis consumer group name
consumerId	auto-generated UUID	Unique consumer identifier
blockTimeoutMs	5000	XREADGROUP block timeout in ms
batchSize	10	Max messages per read

Publish Options

The publish method accepts an optional options object for delayed delivery and metadata headers:

await queue.publish('order.placed', order, {
  delayMs: 30_000, // delay delivery by 30 seconds
  headers: { 'x-correlation-id': traceId }, // metadata
});

Message Shape

Consumers receive a QueueMessage<T> with the following shape:

interface QueueMessage<T = unknown> {
  readonly id: string;
  readonly topic: string;
  readonly payload: T;
  readonly headers: Record<string, string>;
  readonly attempt: number; // starts at 1, increments on retry
  readonly publishedAt: Date;
}

Dead Letter Queue

Messages that exhaust all retries are moved to the dead letter queue. Access them programmatically or via CLI:

const deadLetters = await queue.getDeadLetters('order.placed');
// Re-process or inspect...
await queue.purgeDeadLetters('order.placed');

CLI:

myapp queue:dead-letter:list

Concurrency

The concurrency option on @Consume controls parallel message processing per consumer. The default is 1 (sequential).

@Consume({ topic: 'email.send', name: 'send-email', concurrency: 5 })
async handleSend(message: QueueMessage<Email>): Promise<void> {
  await sendEmail(message.payload);
}

Use 1 for ordering-sensitive work. Higher values (e.g., 3-10) are appropriate for independent I/O-bound tasks like sending emails or calling external APIs.

Serialization

Payloads are serialized with JSON.stringify and deserialized with JSON.parse. Values must be JSON-serializable -- Date instances become ISO strings and class instances lose their prototype chain.

Consumer Middleware

The contract declares QueueConsumerMiddlewarePool — a chain every driver runs around each @Consume handler execution (via the contract's runConsumerPipeline template). This is the seam cross-cutting extensions use to observe or wrap consumption without any driver knowledge:

export interface ConsumeContext {
  readonly message: QueueMessage;
  readonly topic: string;
  readonly consumerName: string;
}

export interface ConsumerMiddleware {
  intercept(
    context: ConsumeContext,
    next: () => Promise<unknown>,
  ): Promise<unknown>;
}

Contribute an implementation class to the pool (pool order = outermost first). A middleware must call next() exactly once and let its rejection propagate — swallowing the error would defeat retries and dead-lettering.

@Module({
  name: 'consume-audit',
  imports: [QueueModule],
  providers: [ConsumeAuditMiddleware],
  pools: [
    {
      pool: QueueConsumerMiddlewarePool,
      key: 'consume-audit',
      useClass: ConsumeAuditMiddleware,
    },
  ],
})
class ConsumeAuditModule {}

Two framework extensions ride this seam:

• @modularityjs/queue-metrics contributes ConsumeMetricsMiddleware — queue.consumed (counter, {topic, consumer, outcome}), queue.consume.duration_ms (histogram), and queue.consume.lag_ms (publish-to-consume latency) — alongside its publish-side @Plugin metrics.
• @modularityjs/queue-telemetry contributes ConsumeTelemetryMiddleware — extracts the W3C trace context its publish-side plugin injected into message headers and opens a CONSUMER span, so the trace continues across the publish/consume boundary with no handler code.

Background Jobs (@modularityjs/queue-jobs)

queue-jobs layers typed background jobs over any queue driver: a job is a class, its payload is typed end-to-end, and dispatching takes the class — no topic strings in application code.

import { Job, JobDefinition } from '@modularityjs/queue-jobs';

interface InvoicePayload {
  invoiceId: string;
}

@JobDefinition({
  name: 'send-invoice',
  retryPolicy: { maxAttempts: 5, strategy: 'exponential', delayMs: 2000 },
  concurrency: 2,
})
class SendInvoiceJob extends Job<InvoicePayload> {
  async handle(payload: InvoicePayload): Promise<void> {
    // constructor-inject services like any provider
  }
}

@JobDefinition applies @Injectable() and registers the class's execute method as a @Consume handler on topic jobs:<name> — retry policy and concurrency are ordinary @Consume options, enforced by whichever driver is active. Register jobs with the jobPoolEntries helper (it contributes each class to both JobsPool and QueueConsumersPool; the classes must also be listed in providers):

@Module({
  name: 'billing',
  imports: [QueueModule, QueueJobsModule],
  providers: [SendInvoiceJob],
  pools: [...jobPoolEntries(SendInvoiceJob)],
})
class BillingModule {}

Dispatch by class through JobDispatcher:

import { Inject, Injectable } from '@modularityjs/di';
import { JobDispatcher } from '@modularityjs/queue-jobs';

@Injectable()
class BillingService {
  constructor(@Inject(JobDispatcher) private readonly jobs: JobDispatcher) {}

  async invoice(invoiceId: string): Promise<void> {
    await this.jobs.dispatch(SendInvoiceJob, { invoiceId });
    // or delayed:
    await this.jobs.dispatch(
      SendInvoiceJob,
      { invoiceId },
      { delayMs: 60_000 },
    );
  }
}

The payload type is checked against the job class at the call site. Dispatching a job that isn't wired via jobPoolEntries throws a StateException naming the fix — a job with no consumer would otherwise vanish into an unconsumed topic. QueueJobsModule has no driver of its own: wire it next to QueueModule and any driver (QueueMemoryModule, QueueRedisModule, QueueRabbitmqModule). See the Jobs wiring recipe for a complete example.