@qified/rabbitmq

RabbitMQ message and task provider for Qified.

This package implements a message provider and a task provider backed by RabbitMQ. The message provider uses queues for publish/subscribe operations, and the task provider adds reliable task queue processing with retries, timeouts, and dead-letter queues.

Table of Contents

• Installation
• Usage with Qified
  • Message Provider
  • Task Provider
• API
  • RabbitMqMessageProviderOptions
  • defaultRabbitMqUri
  • RabbitMqMessageProvider
    • constructor
    • publish
    • subscribe
    • unsubscribe
    • disconnect
  • createQified
  • RabbitMqTaskProviderOptions
  • RabbitMqTaskProvider
    • constructor
    • connect
    • enqueue
    • dequeue
    • unsubscribe
    • disconnect
    • getDeadLetterTasks
    • getQueueStats
    • clearQueue
• Contributing
• License

Installation

pnpm add @qified/rabbitmq

Usage with Qified

Message Provider

import { createQified } from "@qified/rabbitmq";
import type { Message } from "qified";

const qified = createQified({ uri: "amqp://localhost:5672" });

await qified.subscribe("example-topic", {
        async handler(message: Message) {
                console.log(message);
        },
});

await qified.publish("example-topic", { id: "1", data: "Hello from RabbitMQ!" });

await qified.disconnect();

Task Provider

import { RabbitMqTaskProvider } from "@qified/rabbitmq";

const taskProvider = new RabbitMqTaskProvider({ uri: "amqp://localhost:5672" });

// Enqueue a task
await taskProvider.enqueue("my-queue", {
        data: { action: "send-email", to: "[email protected]" },
});

// Dequeue and process tasks
await taskProvider.dequeue("my-queue", {
        id: "email-handler",
        handler: async (task, ctx) => {
                console.log("Processing task:", task.data);

                // Access attempt metadata
                console.log(`Attempt ${ctx.metadata.attempt} of ${ctx.metadata.maxRetries}`);

                // Extend the deadline if needed
                await ctx.extend(10_000);

                // Acknowledge the task on success
                await ctx.ack();
        },
});

// Get queue statistics
const stats = await taskProvider.getQueueStats("my-queue");
console.log(stats); // { waiting, processing, deadLetter }

// Get dead-letter tasks for inspection
const deadLetters = await taskProvider.getDeadLetterTasks("my-queue");

// Clean up
await taskProvider.disconnect();

API

RabbitMqMessageProviderOptions

Configuration options for the RabbitMQ message provider.

• uri?: RabbitMQ connection URI. Defaults to defaultRabbitMqUri.

defaultRabbitMqUri

Default RabbitMQ connection string ("amqp://localhost:5672").

RabbitMqMessageProvider

Implements the MessageProvider interface using RabbitMQ queues.

constructor(options?: RabbitMqMessageProviderOptions)

Creates a new provider.

Options:

• uri: RabbitMQ connection URI (defaults to "amqp://localhost:5672").

publish(topic: string, message: Message)

Publishes a message to a topic.

subscribe(topic: string, handler: TopicHandler)

Subscribes a handler to a topic.

unsubscribe(topic: string, id?: string)

Unsubscribes a handler by id or all handlers for a topic.

disconnect()

Cancels all subscriptions and closes the underlying RabbitMQ connection.

createQified(options?: RabbitMqMessageProviderOptions)

Convenience factory that returns a Qified instance configured with RabbitMqMessageProvider.

RabbitMqTaskProviderOptions

Configuration options for the RabbitMQ task provider. Extends TaskProviderOptions.

• uri?: RabbitMQ connection URI. Defaults to "amqp://localhost:5672".
• id?: Unique identifier for this provider instance. Defaults to "@qified/rabbitmq-task".
• timeout?: Default timeout in milliseconds for task processing. Defaults to 30000.
• retries?: Default maximum retry attempts before a task is moved to the dead-letter queue. Defaults to 3.
• reconnectTimeInSeconds?: Time in seconds to wait before reconnecting after connection loss. Set to 0 to disable. Defaults to 5.

RabbitMqTaskProvider

Implements the TaskProvider interface using RabbitMQ durable queues for reliable task processing. Extends Hookified for event emission. Features include:

• Automatic retries with configurable max attempts
• Task timeouts with automatic rejection on expiry
• Dead-letter queue for failed tasks
• Automatic reconnection on connection loss

constructor(options?: RabbitMqTaskProviderOptions)

Creates a new task provider instance.

connect()

Explicitly connects to RabbitMQ. Called automatically on first enqueue or dequeue if not called manually.

enqueue(queue: string, taskData: EnqueueTask)

Enqueues a task to the specified queue. Returns a Promise<string> with the generated task ID.

Task data options:

• data: The task payload (any serializable value).
• id?: Custom task ID. Auto-generated if omitted.
• timeout?: Per-task timeout override in milliseconds.
• maxRetries?: Per-task max retry override.
• priority?: Task priority value.

dequeue(queue: string, handler: TaskHandler)

Registers a handler to process tasks from the specified queue. The handler receives a Task and a TaskContext.

TaskContext methods:

• ack(): Acknowledge the task (removes it from the queue).
• reject(requeue?: boolean): Reject the task. If requeue is true (default), re-enqueues for retry. After max retries, moves to dead-letter queue.
• extend(ms: number): Extend the processing deadline by the given milliseconds.
• metadata: Object with { attempt, maxRetries } for the current task.

unsubscribe(queue: string, id?: string)

Removes a handler by id, or all handlers for the queue if no id is provided.

disconnect(force?: boolean)

Disconnects from RabbitMQ and cleans up all consumers, timers, and in-memory state. If force is true, skips graceful channel close.

getDeadLetterTasks(queue: string)

Returns an array of tasks that have been moved to the dead-letter queue for the given queue.

getQueueStats(queue: string)

Returns statistics for the given queue:

{ waiting: number; processing: number; deadLetter: number }

clearQueue(queue: string)

Purges all tasks from the queue and its dead-letter queue, and clears all in-memory tracking state.

Contributing

Contributions are welcome! Please read the CONTRIBUTING.md and CODE_OF_CONDUCT.md for details on our process.

License

MIT © Jared Wray. See LICENSE for details.