RabbitMQ

@opentf/esrun-rabbitmq is a RabbitMQ client speaking AMQP 0-9-1. It is written in JavaScript over runtime:net, so it needs no grant beyond reaching the broker.

npm install @opentf/esrun-rabbitmq

esrun --allow-net=rabbitmq.example.com app.js

Connect

openConnection takes an amqp:// URL, dials it and finishes the AMQP handshake. Everything the connection and its channels do happens beneath it; the default exchange ("") routes by queue name, so a first message needs no topology.

import { openConnection } from "@opentf/esrun-rabbitmq";

const connection = await openConnection("amqp://guest:guest@localhost/");
const channel = await connection.channel();

await channel.queueDeclare("jobs", { durable: true });
await channel.publish("", "jobs", "hello");

amqps: is TLS from the first byte. The URL carries the credentials and the virtual host; any of them can be overridden in the second argument, which also holds the heartbeat, timeouts and limits. RabbitMQ's URL query parameters (?heartbeat=30&connection_timeout=5000) are read too. For a broker behind a private certificate authority, pass its PEM as ca:

import { file } from "runtime:fs";

const connection = await openConnection("amqps://app:secret@rabbit.internal/prod", {
  ca: await file("certs/ca.pem").text(),
});

The package reference lists them all.

Declare topology

Exchanges, queues and the bindings between them are declared on a channel. A declaration is idempotent when it matches the existing one and a 406 when it does not, which is what makes declaring at startup safe.

await channel.exchangeDeclare("events", "topic", { durable: true });

const { queue } = await channel.queueDeclare("", { exclusive: true });
await channel.queueBind(queue, "events", "user.*");

await channel.publish("events", "user.created", payload, {
  contentType: "application/json",
  deliveryMode: 2, // persistent
});

An empty queue name asks the broker to generate one; the declaration answers with it. arguments carries broker extensions such as x-message-ttl or a dead-letter policy.

Confirm your publishes

By default publish resolves once the frames are written, which is all the protocol promises. confirmSelect() makes the broker acknowledge every publish, and then publish resolves only once it has — so awaiting the publish awaits the message's fate. A message the broker cannot take rejects with ERR_RABBITMQ_NACK, and one it has not confirmed within the publish's timeout (30 seconds unless set) rejects with ERR_RABBITMQ_TIMEOUT. While the broker blocks publishing during a resource alarm, publishes wait for it to lift, within the same timeout.

await channel.confirmSelect();
await channel.publish("events", "user.created", payload); // resolves once RabbitMQ has it

Consume a queue

subscribe returns an async iterable of deliveries; each carries its own ack, nack and reject. prefetch bounds how many are unacknowledged at once; without it a subscription takes 250.

import { subscribe } from "@opentf/esrun-rabbitmq";

const subscription = await subscribe(channel, "jobs", { prefetch: 16 });
for await (const delivery of subscription) {
  try {
    await handle(JSON.parse(new TextDecoder().decode(delivery.body)));
    await delivery.ack();
  } catch (e) {
    await delivery.nack({ requeue: true }); // or reject it into a dead-letter queue
  }
}

Breaking out of the for await cancels the consumer; so does await subscription.cancel(). Deliveries that had arrived but were not read go back to the queue for another consumer. If the broker cancels it — a queue deleted under the consumer — the loop ends and subscription.cancellation says the broker did it and why.

Request/reply

RabbitMQ's direct reply-to gives a private reply address on the calling channel, so an RPC needs no queue. RpcClient consumes it and matches answers by correlation id.

import { RpcClient } from "@opentf/esrun-rabbitmq";

const rpc = await RpcClient.open(await connection.channel());
const answer = await rpc.call("fib", "10", { timeout: 5_000 });
console.log(new TextDecoder().decode(answer.body));

await rpc.close();

The responder consumes its queue as usual, and publishes its answer to the replyTo address the broker put on the request, echoing its correlationId. Concurrent calls on one channel are told apart by id. A call nobody answers rejects with ERR_RABBITMQ_TIMEOUT.

Survive a broker restart

A long-running consumer or publisher should not end when its connection does. openSession gives you channels that are reopened — and set up again — after a lost connection, and subscriptions that keep yielding across it.

import { openSession } from "@opentf/esrun-rabbitmq";

const session = await openSession("amqp://app:secret@rabbit.internal/");
session.onDisconnect = (error) => console.warn("lost the broker:", error.message);

const jobs = await session.channel({
  confirm: true,
  setup: (channel) => channel.queueDeclare("jobs", { durable: true }),
});

for await (const delivery of jobs.subscribe("jobs", { prefetch: 16 })) {
  await handle(delivery);
  await delivery.ack();
}

Put everything the channel needs in setup: it runs again after every reconnect. A message you were handling when the connection dropped is delivered again (delivery.redelivered is true), so make the handler safe to repeat. A publish whose confirm had not arrived when the connection dropped rejects rather than being sent twice; publish it again if it must get through.

Pool connections

A connection is one socket, so a burst of publishing shares it. A Pool spreads that over several, and dials a replacement when one dies. Re-establishing what you declared after a loss is yours to do in onConnect.

import { Pool } from "@opentf/esrun-rabbitmq";

const pool = new Pool("amqp://localhost/", {
  size: 4,
  onConnect: async (connection) => {
    const channel = await connection.channel();
    await channel.queueDeclare("jobs", { durable: true });
    await channel.close();
  },
});

await pool.withChannel((channel) => channel.publish("", "jobs", "hello"));
await pool.close();

withChannel borrows a connection, opens a channel, runs your function and closes it. For something long-lived — a subscription — borrow whole with pool.acquire() and pool.release(connection).

Handle failures

Every failure is a RabbitError carrying the broker's own replyCode and replyText. soft is what is still usable: a channel exception (a queue that does not exist, a failed precondition) kills the channel and leaves the connection standing, while a connection exception kills everything.

import { RabbitError } from "@opentf/esrun-rabbitmq";

try {
  await channel.publish("events", "user.created", payload);
} catch (e) {
  if (e instanceof RabbitError && e.soft) await republishOnANewChannel(payload);
  else throw e;
}

A lost socket is fatal and latched: a later frame cannot be trusted, so the connection does not try to resynchronise. A pool turns that into a new connection on the next borrow.

Test locally

docker run -d --name rabbitmq -p 5672:5672 rabbitmq:4-alpine

const connection = await openConnection("amqp://guest:guest@127.0.0.1:5672/");

How the client is put together — the framing, why writes are serialised, and what the pool's recovery does and does not do: Internals: RabbitMQ.