RabbitMQBroker

RabbitMQBroker is the main entry point for Rabbit Relay.

It creates queue/exchange interfaces, publishes events, consumes events, exposes lifecycle hooks, and provides operational helpers such as health checks, topology planning, topology validation, and DLQ redrive.

Creating a broker

import { RabbitMQBroker } from "@bitspacerlabs/rabbit-relay";

const broker = new RabbitMQBroker("orders-service");

The first argument is the peer/service name. It is used in lifecycle events, health output, and operational visibility.

Broker configuration

const broker = new RabbitMQBroker("orders-service", {
  connectionUrl: "amqp://user:password@rabbitmq.internal:5672",
  connectionName: "orders-service-primary",
  shutdownTimeoutMs: 30_000,
  exchangeType: "topic",
  routingKey: "#",
  durable: true,
  publisherConfirms: false,
  topologyMode: "assert",
  maxMessageBytes: 256 * 1024,
});

Broker-only options configure connection and shutdown ownership:

The remaining broker-level options become exchange defaults.

Exchange-level options override broker-level options.

Each broker owns its connection and channels. Calling close() does not close resources owned by other broker instances.

Creating a queue/exchange interface

const sub = await broker
  .queue("orders.q")
  .exchange("orders.ex", {
    exchangeType: "topic",
    routingKey: "orders.*",
  });

This creates a broker interface bound to:

• one queue
• one exchange
• one routing key pattern

By default, Rabbit Relay asserts the exchange, queue, and binding.

Creating an exchange-only interface (publisher)

For producers that only publish and don't consume, use broker.exchange() to declare just the exchange — no queue, no binding, no consumer. This keeps topology ownership clean in multi-process setups: the producer owns the exchange; the consumer owns the queue and bindings.

const pub = await broker.exchange("orders.ex", {
  exchangeType: "topic",
});

await pub.produce(eventFactory(data));

The plan for an exchange-only interface contains queues: [] and bindings: []:

pub.planTopology();
// {
//   exchanges: [{ name: "orders.ex", type: "topic", ... }],
//   queues: [],
//   bindings: [],
// }

Notes:

• produce(), publish(), and request() are fully available on an exchange-only interface.
• consume() is not available — it throws. Use .queue(...).exchange(...) to consume.
• handle() is not useful here (there is no consumer to dispatch to); register handlers on the consumer's queue/exchange interface instead.

Exchange options

const sub = await broker
  .queue("orders.q")
  .exchange("orders.ex", {
    exchangeType: "topic",
    routingKey: "orders.*",
    durable: true,
    publisherConfirms: true,
    topologyMode: "assert",
    passiveQueue: false,
    maxMessageBytes: 256 * 1024,
    queueArgs: {
      "x-queue-type": "quorum",
    },
    deadLetter: {
      exchange: "orders.dlx",
      queue: "orders.dlq",
      routingKey: "orders.dead",
      autoDeclare: true,
    },
    amqp: {
      exchange: {},
      queue: {},
      bind: {},
    },
  });

Topology modes

topologyMode controls what Rabbit Relay does when .exchange(...) is called.

const sub = await broker
  .queue("orders.q")
  .exchange("orders.ex", {
    exchangeType: "topic",
    routingKey: "orders.*",
    topologyMode: "passive",
  });

Supported values:

Default:

topologyMode: "assert"

Use topologyMode: "passive" for infrastructure-managed topology.

Use topologyMode: "plan-only" when you want to inspect topology without creating RabbitMQ resources.

See Topology Modes.

passiveQueue compatibility

passiveQueue is still supported for backward compatibility.

const sub = await broker
  .queue("orders.q")
  .exchange("orders.ex", {
    passiveQueue: true,
  });

passiveQueue only affects the main queue declaration behavior.

For new infrastructure-managed deployments, prefer:

topologyMode: "passive"

This makes ownership clearer because it applies to topology behavior as a whole.

Producing events

await sub.produce(orderCreated({ id: "o-1" }));

You can publish multiple events sequentially. produceMany() is deprecated; use the variadic produce(...events) instead:

await sub.produce(
  orderCreated({ id: "o-1" }),
  orderCreated({ id: "o-2" })
);

Use publish() when you need per-message options:

await sub.publish(orderCreated({ id: "o-1" }), {
  routingKey: "orders.created",
  maxMessageBytes: 256 * 1024,
  amqp: {
    publish: {
      persistent: true,
      priority: 5,
    },
  },
});

Consuming events

sub.handle("order.created", async (_id, event) => {
  console.log(event.data);
});

await sub.consume({
  prefetch: 10,
  concurrency: 5,
});

Supported consume options include:

Delayed retry topology

When delayed retry is configured, Rabbit Relay may need retry exchanges and retry queues.

await sub.consume({
  onError: "retry",
  retry: {
    attempts: 3,
    delayMs: 5000,
    then: "dead-letter",
  },
});

Delayed retry follows topologyMode:

With retry.backoff: "exponential", one parking exchange/queue pair is declared per attempt (<queue>.retry.aN.exchange / <queue>.retry.aN.<ttl>.queue), each with its own TTL. In "passive" mode every attempt pair must already exist.

RPC request

Use request() when you want a typed reply.

const reply = await sub.request<{ ok: boolean }>(
  orderCreated({ id: "o-1" }),
  { timeoutMs: 5000 }
);

Middleware

sub.use(async (ctx, next) => {
  console.log(ctx.queue, ctx.event.name);
  await next();
});

Middleware wraps handler execution for one broker interface.

Lifecycle hooks

sub.on("consumer.started", (event) => {
  console.log(event.queue, event.prefetch, event.concurrency);
});

Broker-level hooks are also supported:

broker.on("broker.closed", (event) => {
  console.log(event.peerName);
});

Topology planner

planTopology() returns Rabbit Relay's known topology plan.

It does not contact RabbitMQ.

const plan = sub.planTopology();

Broker-level plan:

const fullPlan = broker.planTopology();

See Topology Planner.

Topology validation

validateTopology() checks planned exchanges and queues using passive AMQP checks.

It does not declare or modify RabbitMQ resources.

const result = await sub.validateTopology();

if (!result.valid) {
  console.error(result.issues);
}

Broker-level validation:

const result = await broker.validateTopology();

See Topology Validation.

DLQ redrive

const result = await broker.redriveDlq({
  fromQueue: "orders.dlq",
  toExchange: "orders.ex",
  routingKey: "orders.created",
  limit: 100,
  dryRun: true,
});

The original DLQ message is acknowledged only after republish succeeds.

Health

const health = await broker.health();

Health includes connection/channel state and consumer state.

withChannel escape hatch

Use withChannel() for advanced amqplib operations.

await broker.withChannel(async (channel) => {
  const info = await channel.checkQueue("orders.q");
  console.log(info.messageCount);
});

Close

await broker.close();

This stops active consumers, closes Rabbit Relay resources, and clears lifecycle hooks.

Summary

• RabbitMQBroker is the main runtime API
• .queue(...).exchange(...) creates a typed broker interface
• topologyMode controls topology ownership behavior
• planTopology() is read-only
• validateTopology() is passive and non-mutating
• redriveDlq() safely republishs DLQ messages