amqp-petite-connector

A small, petite and opinionated way to connect to AMQP (RabbitMQ). Built for ease of use with a small footprint — follow a set of conventions and it stays out of your way.

Wraps amqplib with a clean declarative API for the common use cases.

Install

npm install amqp-petite-connector

Quick start

const petite = require("amqp-petite-connector")

const exchanges = [{
  name: "user-events",
  type: "fanout",
  options: { durable: true }
}]

const queues = [{
  name: "process-user-created",
  binding: "user-events",
  callback: async (msg) => {
    const data = JSON.parse(msg.content.toString())
    // process the message
    // if this throws without DLX: message is acked and dropped
    // if this throws with DLX: message is nacked and retried
  }
}]

await petite.connect(queues, exchanges, {
  connectionString: "amqp://localhost"  // or set RABBIT_URL env var
})

petite.publish("user-events", { event: "user.created", userId: 123 })

API

connect(queues, exchanges, options?)

Connects to AMQP, asserts exchanges and queues, starts consuming. Throws on connection failure. Auto-reconnects with exponential backoff on connection loss after initial connect.

Options:

| Option | Type | Default | Description | |--------|------|---------|-------------| | connectionString | string | RABBIT_URL env | AMQP connection string | | prefetch | number | - | Channel prefetch count (QoS) | | reconnect | boolean | true | Auto-reconnect on connection loss | | heartbeat | number | - | AMQP heartbeat interval in seconds | | logger | object | console | Custom logger with log() and error() methods |

publish(exchange, data, options?)

Publishes JSON-serialized data to an exchange. Returns false if the channel buffer is full (backpressure). Throws if not connected, if data is undefined, or if publishing fails.

Options:

| Option | Type | Default | Description | |--------|------|---------|-------------| | routingKey | string | "" | Routing key for topic/direct exchanges | | persistent | boolean | true | Message persistence | | headers | object | - | Custom message headers | | expiration | string | - | Message TTL in milliseconds | | correlationId | string | - | Correlation ID for RPC patterns | | replyTo | string | - | Reply-to queue for RPC patterns | | priority | number | - | Message priority (0-9) |

Any additional options are passed through to amqplib's channel.publish().

disconnect(options?)

Gracefully shuts down: cancels consumers, optionally waits for in-flight handlers, closes channel and connection. Does not trigger auto-reconnect.

Returns { drained: boolean, remaining: number }.

Options:

| Option | Type | Default | Description | |--------|------|---------|-------------| | drainTimeoutMs | number | - | Wait up to this long for in-flight message handlers to finish before closing the channel. Without it, disconnect closes immediately (back-compat). |

// SIGTERM handler — wait up to 20s for in-flight handlers
const { drained, remaining } = await petite.disconnect({ drainTimeoutMs: 20_000 })
if (!drained) logger.warn(`forced disconnect with ${remaining} in-flight`)

When choosing drainTimeoutMs, account for your container orchestrator's stop timeout. Common use is docker stop -t 30, so anything above ~25s gets cut short by SIGKILL.

getConnection()

Returns the underlying amqplib connection for advanced use cases. Returns null if not connected.

Contracts

Exchange definition

{
  name: string,          // Exchange name
  type: string,          // "fanout" | "topic" | "direct" | "headers"
  options: {...}          // amqplib exchange options (durable, autoDelete, etc.)
}

Queue definition

{
  name: string,          // Queue name
  binding: string,       // Exchange name to bind to
  routingKey: string,    // Optional. Routing key for topic/direct exchanges
  callback: async (msg, channel) => {},  // Message handler
  options: {...},        // Optional. Default: { durable: true }
  dlx: {                 // Optional. Dead letter exchange config
    messageTtl: number,  // Optional. Retry delay in ms. Default: 300000 (5min)
    maxRetries: number,  // Optional. Max retries before dead queue. Default: 5
    deadQueue: boolean   // Optional. Create dead queue. Default: true
  }
}

Opinions

• Messages are always JSON-serialized
• Messages are persistent by default
• Consumers are expected to be idempotent
• Exchanges are events, queues are commands
• Without DLX: failed messages are acked and dropped (no blocking)
• With DLX: failed messages retry up to maxRetries (default 5) times, then go to the dead queue
• Connection auto-recovers with exponential backoff + jitter
• Exchange assertion failures propagate immediately (fail fast)
• Singleton connection — one connection per process

Dead letter exchange (DLX)

When dlx is set on a queue, three resources are created automatically:

• {name}_dlx — fanout exchange for dead letters
• {name}_retry — queue with TTL, republishes to the original exchange
• {name}_dead — queue for messages that failed >= maxRetries times

Failed messages (unhandled exceptions) are nacked and routed to the retry queue. After the TTL expires, they're republished to the original exchange. After maxRetries failures, they land on the dead queue. Monitor the dead queue length.

Disable the dead queue with deadQueue: false. If you don't want DLX at all, just catch your exceptions.

Topic exchange example

const exchanges = [{
  name: "events",
  type: "topic",
  options: { durable: true }
}]

const queues = [{
  name: "audit-log",
  binding: "events",
  routingKey: "user.*",
  callback: async (msg) => {
    const data = JSON.parse(msg.content.toString())
    await logAuditEvent(data)
  }
}]

await petite.connect(queues, exchanges)

petite.publish("events", { userId: 123 }, { routingKey: "user.created" })
petite.publish("events", { userId: 123 }, { routingKey: "user.deleted" })

Headers example

petite.publish("events", { orderId: 456 }, {
  routingKey: "order.completed",
  headers: {
    "x-source": "checkout-service",
    "x-trace-id": "abc-123"
  }
})

DLX example

const queues = [{
  name: "process-orders",
  binding: "order-events",
  dlx: {
    messageTtl: 30000,  // retry after 30 seconds
    maxRetries: 3        // dead-letter after 3 failures
  },
  callback: async (msg) => {
    const order = JSON.parse(msg.content.toString())
    await processOrder(order)
    // throw here to see dead lettering in action
  }
}]

Connection options example

await petite.connect(queues, exchanges, {
  connectionString: "amqp://user:pass@broker:5672",
  prefetch: 200,
  heartbeat: 60,
  reconnect: true,
  logger: myCustomLogger  // must have log() and error()
})

Graceful shutdown example

Wire disconnect({ drainTimeoutMs }) into your SIGTERM handler so in-flight message handlers get a chance to finish before the channel closes. Without this, the broker redelivers in-flight messages on the next worker — fine if your handlers are idempotent, painful if they aren't.

process.on("SIGTERM", async () => {
  const { drained, remaining } = await petite.disconnect({ drainTimeoutMs: 20_000 })
  if (!drained) {
    logger.warn(`forced disconnect with ${remaining} in-flight handlers`)
  }
  process.exit(0)
})

Error handling

try {
  await petite.connect(queues, exchanges, options)
} catch (err) {
  // Initial connection failed — handle or retry yourself
}

try {
  petite.publish("exchange", data)
} catch (err) {
  // Not connected, or channel error — message was not sent
}

TypeScript

Full type definitions are included. Import types directly:

import petite, {
  QueueDefinition,
  ExchangeDefinition,
  PublishOptions,
  DisconnectOptions,
  DisconnectResult,
} from "amqp-petite-connector"

Requirements

Node.js >= 18