Packages

@xtaskjs/queues

Transport-agnostic queue handlers, broker helpers, publish decorators, and lifecycle-managed consumers.

npm install @xtaskjs/queues reflect-metadata Package path: packages/queues

Overview

What this package owns in the runtime

Queues brings broker-backed and in-memory message delivery into xtaskjs. It registers named transports, discovers decorated consumers from DI-managed services, exposes injectable producer APIs, and coordinates retries, dead-letter routing, and consumer start-stop behavior through the application lifecycle.

What it provides

configureQueues() sets default transport behavior and can disable the automatic in-memory transport.
registerQueueTransport(), registerInMemoryQueueTransport(), createRabbitMqTransport(), and createMqttTransport() connect brokers or local transports.
QueueHandler(), QueuePattern(), PublishToQueue(), InjectQueueService(), and InjectQueueTransport() cover consumer, producer, and transport-injection flows.
QueueService can publish messages, create producers, inspect consumers and groups, and start or stop queue runtime state through lifecycle-managed APIs.

How it fits

Initialized automatically by @xtaskjs/core when the package is installed and queue configuration or decorated consumers are present.
Pairs with the built-in in-memory transport for local development and tests, and with RabbitMQ or MQTT helpers for broker-backed delivery.
Demonstrated by the 16-queues_memory_app and 17-queues_rabbitmq_app samples.

Usage Chart

How strongly this package shapes the runtime

Bootstrap

4/5

Dependency Injection

4/5

Messaging

5/5

Operations

5/5

Integrations

4/5

Package Flow

How this package moves through xtaskjs runtime phases

Before startup

Configure queue defaults, register transports, and decorate consumers while modules load so the queue runtime knows which producers and handlers to wire.

During CreateApplication()

The queues lifecycle publishes QueueService and named transport tokens into the container, discovers decorated consumers, and prepares them for lifecycle-ready startup.

During app.close()

Active consumers are stopped and connected transports are disconnected automatically so broker resources shut down with the application.

API Surface

Representative exports from the upstream package

Configuration and transports

configureQueues
registerQueueTransport
registerInMemoryQueueTransport
createRabbitMqTransport
createMqttTransport

Decorators and injectors

QueueHandler
QueueSubscribe
QueuePattern
PublishToQueue
InjectQueueService
InjectQueueLifecycleManager
InjectQueueTransport

Runtime service and lifecycle

QueueService
initializeQueueIntegration
shutdownQueueIntegration
getQueueServiceToken
getQueueTransportToken

Usage

Typical adoption flow

1. Configure transports and defaults

Start with the built-in in-memory transport for local workflows, or register RabbitMQ or MQTT helpers when messages must cross process boundaries.

2. Decorate consumers and publishers

Use QueueHandler and QueuePattern for consumers, PublishToQueue for method-result events, and InjectQueueService when services need direct publish or producer APIs.

3. Inspect and control the queue runtime

Use QueueService to list groups and consumers, create producers with defaults, and start or stop queue processing for diagnostics or operations endpoints.

Example

Reference snippet

Named transport plus decorated queue consumers

import { Service } from "@xtaskjs/core";
import {
  InjectQueueService,
  PublishToQueue,
  QueueHandler,
  QueueService,
  configureQueues,
  registerInMemoryQueueTransport,
} from "@xtaskjs/queues";

configureQueues({
  defaultTransportName: "memory",
  autoCreateDefaultInMemoryTransport: false,
});

registerInMemoryQueueTransport({
  name: "memory",
  kind: "in-memory",
});

@Service()
export class OrdersQueueService {
  constructor(
    @InjectQueueService()
    private readonly queues: QueueService
  ) {}

  async publishOrder(orderId: string) {
    await this.queues.publish("orders.created", { orderId });
  }

  @PublishToQueue("orders.completed", { transportName: "memory" })
  completeOrder(orderId: string) {
    return { orderId, completedAt: new Date().toISOString() };
  }

  @QueueHandler("orders.created", { name: "orders.created", transportName: "memory" })
  onOrderCreated(payload: { orderId: string }) {
    console.log("received", payload.orderId);
  }
}

Samples

Official samples to inspect next

Reference samples: 16-queues_memory_app and 17-queues_rabbitmq_app