@probitas/client-rabbitmq

RabbitMQ client for Probitas scenario testing framework.

This package provides a RabbitMQ client designed for integration testing of message-driven applications.

Features

• Queue Operations: Declare, bind, purge, and delete queues
• Exchange Operations: Declare and delete exchanges (direct, topic, fanout, headers)
• Publishing: Publish messages with routing keys and headers
• Consuming: Consume messages with acknowledgment support
• Resource Management: Implements AsyncDisposable for proper cleanup

Installation

deno add jsr:@probitas/client-rabbitmq

Quick Start

import { createRabbitMqClient } from "@probitas/client-rabbitmq";

// Using string URL
const client = await createRabbitMqClient({
  url: "amqp://localhost:5672",
});

// Or using connection config object
const client2 = await createRabbitMqClient({
  url: {
    host: "localhost",
    port: 5672,
    username: "guest",
    password: "guest",
    vhost: "/",
  },
});

// Create a channel
const channel = await client.channel();

// Declare a queue
await channel.assertQueue("test-queue", { durable: true });

// Publish a message
const content = new TextEncoder().encode("Hello, World!");
await channel.sendToQueue("test-queue", content, {
  contentType: "text/plain",
});

// Consume messages
for await (const msg of channel.consume("test-queue")) {
  console.log("Received:", new TextDecoder().decode(msg.content));
  await channel.ack(msg);
  break;
}

await client.close();

Exchange and Binding

// Declare an exchange
await channel.assertExchange("events", "topic", { durable: true });

// Declare a queue and bind to exchange
await channel.assertQueue("user-events");
await channel.bindQueue("user-events", "events", "user.*");

// Publish to exchange with routing key
const content = new TextEncoder().encode(JSON.stringify({ id: 1, name: "Alice" }));
await channel.publish("events", "user.created", content, {
  contentType: "application/json",
});

Using with using Statement

await using client = await createRabbitMqClient({ url: "amqp://localhost:5672" });
const channel = await client.channel();

await channel.assertQueue("test");
// Client automatically closed when block exits

Related Packages

Package	Description
@probitas/client	Core utilities and types
@probitas/client-sqs	AWS SQS client

Links

• GitHub Repository
• Probitas Framework
• RabbitMQ

deno add jsr:@probitas/client-rabbitmq

class RabbitMqChannelError extends RabbitMqError

new RabbitMqChannelError(message: string, options?: RabbitMqChannelErrorOptions)

class RabbitMqConnectionError extends RabbitMqError

new RabbitMqConnectionError(message: string, options?: RabbitMqErrorOptions)

class RabbitMqError extends ClientError

new RabbitMqError(message: string, _: unknown, options?: RabbitMqErrorOptions)

class RabbitMqNotFoundError extends RabbitMqError

new RabbitMqNotFoundError(message: string, options: RabbitMqNotFoundErrorOptions)

class RabbitMqPreconditionFailedError extends RabbitMqError

new RabbitMqPreconditionFailedError(message: string, options: RabbitMqPreconditionFailedErrorOptions)

interface RabbitMqAckResult

interface RabbitMqChannel extends AsyncDisposable

assertExchange(name: string, type: RabbitMqExchangeType, options?: RabbitMqExchangeOptions): Promise<RabbitMqExchangeResult>

deleteExchange(name: string, options?: CommonOptions): Promise<RabbitMqExchangeResult>

assertQueue(name: string, options?: RabbitMqQueueOptions): Promise<RabbitMqQueueResult>

deleteQueue(name: string, options?: CommonOptions): Promise<RabbitMqQueueResult>

purgeQueue(name: string, options?: CommonOptions): Promise<RabbitMqQueueResult>

bindQueue(queue: string, exchange: string, routingKey: string, options?: CommonOptions): Promise<RabbitMqExchangeResult>

unbindQueue(queue: string, exchange: string, routingKey: string, options?: CommonOptions): Promise<RabbitMqExchangeResult>

publish(exchange: string, routingKey: string, content: Uint8Array, options?: RabbitMqPublishOptions): Promise<RabbitMqPublishResult>

sendToQueue(queue: string, content: Uint8Array, options?: RabbitMqPublishOptions): Promise<RabbitMqPublishResult>

get(queue: string, options?: CommonOptions): Promise<RabbitMqConsumeResult>

consume(queue: string, options?: RabbitMqConsumeOptions): AsyncIterable<RabbitMqMessage>

ack(message: RabbitMqMessage, options?: CommonOptions): Promise<RabbitMqAckResult>

nack(message: RabbitMqMessage, options?: RabbitMqNackOptions): Promise<RabbitMqAckResult>

reject(message: RabbitMqMessage, requeue?: boolean): Promise<RabbitMqAckResult>

prefetch(count: number): Promise<void>

close(): Promise<void>

interface RabbitMqChannelErrorOptions extends RabbitMqErrorOptions

interface RabbitMqClient extends AsyncDisposable

channel(): Promise<RabbitMqChannel>

close(): Promise<void>

interface RabbitMqClientConfig extends CommonOptions

interface RabbitMqConnectionConfig extends CommonConnectionConfig

interface RabbitMqConsumeOptions extends CommonOptions

interface RabbitMqConsumeResult

interface RabbitMqErrorOptions extends ErrorOptions

interface RabbitMqExchangeOptions extends CommonOptions

interface RabbitMqExchangeResult

interface RabbitMqMessage

interface RabbitMqMessageFields

interface RabbitMqMessageProperties

interface RabbitMqNackOptions extends CommonOptions

interface RabbitMqNotFoundErrorOptions extends RabbitMqErrorOptions

interface RabbitMqPreconditionFailedErrorOptions extends RabbitMqErrorOptions

interface RabbitMqPublishOptions extends CommonOptions

interface RabbitMqPublishResult

interface RabbitMqQueueOptions extends CommonOptions

interface RabbitMqQueueResult

async function createRabbitMqClient(config: RabbitMqClientConfig): Promise<RabbitMqClient>

const rabbit = await createRabbitMqClient({
  url: "amqp://guest:guest@localhost:5672",
});

const channel = await rabbit.channel();
await channel.assertQueue("my-queue", { durable: true });

const content = new TextEncoder().encode(JSON.stringify({ type: "ORDER" }));
await channel.sendToQueue("my-queue", content, { persistent: true });

await channel.close();
await rabbit.close();

const rabbit = await createRabbitMqClient({
  url: {
    host: "localhost",
    port: 5672,
    username: "guest",
    password: "guest",
    vhost: "/",
  },
});

// Create exchange and queue
await channel.assertExchange("events", "topic", { durable: true });
await channel.assertQueue("user-events");
await channel.bindQueue("user-events", "events", "user.*");

// Publish to exchange
await channel.publish("events", "user.created", content);

for await (const message of channel.consume("my-queue")) {
  console.log("Received:", new TextDecoder().decode(message.content));
  await channel.ack(message);
}

const result = await channel.get("my-queue");
if (result.message) {
  await channel.ack(result.message);
}

await using rabbit = await createRabbitMqClient({
  url: "amqp://localhost:5672",
});

const channel = await rabbit.channel();
// Client automatically closed when scope exits

type RabbitMqExchangeType = "direct" | "topic" | "fanout" | "headers"

type RabbitMqResult = RabbitMqPublishResult | RabbitMqConsumeResult | RabbitMqAckResult | RabbitMqQueueResult | RabbitMqExchangeResult