amqp-resilient

amqp-resilient

Production-ready AMQP client for Node.js with built-in resilience patterns.

Features

• Auto-Reconnection - Exponential backoff with jitter for stable reconnects
• Circuit Breaker - Prevent cascading failures with configurable thresholds
• Retry with DLQ - Failed messages retry with backoff, then route to Dead Letter Queue
• Publisher Confirms - Guaranteed message delivery with confirmation
• Health Monitoring - Built-in health checks for all connections
• TypeScript - Full type safety with generics support
• ESM - Native ES modules support
• Zero Dependencies - Only amqplib as peer dependency

Installation

npm install amqp-resilient amqplib

Quick Start

Connection

import { ConnectionManager } from 'amqp-resilient';

const connection = new ConnectionManager({
  url: 'amqp://localhost:5672',
  connectionName: 'my-service',
  logger: console, // optional
});

await connection.connect();

Consumer

import { BaseConsumer, type MessageContext } from 'amqp-resilient';

interface OrderMessage {
  orderId: string;
  amount: number;
}

class OrderConsumer extends BaseConsumer<OrderMessage> {
  constructor(connection: ConnectionManager) {
    super(connection, {
      queue: 'orders',
      exchange: 'orders.exchange',
      routingKeys: ['order.created', 'order.updated'],
      prefetch: 10,
      maxRetries: 3,
    });
  }

  protected async handle(message: OrderMessage, context: MessageContext): Promise<void> {
    console.log(`Processing order ${message.orderId}`);
    // Throw an error to trigger retry
  }
}

const consumer = new OrderConsumer(connection);
await consumer.start();

Publisher

import { BasePublisher } from 'amqp-resilient';

const publisher = new BasePublisher(connection, {
  exchange: 'orders.exchange',
  confirm: true,
});

const result = await publisher.publish('order.created', {
  orderId: '123',
  amount: 99.99,
});

if (result.success) {
  console.log(`Published message ${result.messageId}`);
}

// Or throw on failure
await publisher.publishOrThrow('order.created', { orderId: '456' });

Configuration

ConnectionManager

Option	Type	Default	Description
url	string	-	Full AMQP URL (amqp://host:port)
host	string	-	RabbitMQ host (alternative to url)
port	number	5672	RabbitMQ port
username	string	guest	Username
password	string	guest	Password
vhost	string	/	Virtual host
connectionName	string	required	Name for logging and health checks
prefetch	number	10	Default channel prefetch count
heartbeat	number	60	Heartbeat interval in seconds
maxReconnectAttempts	number	0	Max reconnect attempts (0 = unlimited)
initialReconnectDelay	number	1000	Initial reconnect delay (ms)
maxReconnectDelay	number	30000	Max reconnect delay (ms)
logger	AmqpLogger	noop	Logger instance

BaseConsumer

Option	Type	Default	Description
queue	string	required	Queue name
exchange	string	required	Exchange to bind
routingKeys	string[]	required	Routing keys to bind
prefetch	number	10	Consumer prefetch count
maxRetries	number	3	Max retries before DLQ
initialRetryDelay	number	1000	Initial retry delay (ms)
maxRetryDelay	number	30000	Max retry delay (ms)
useCircuitBreaker	boolean	true	Enable circuit breaker
exchangeType	ExchangeType	topic	Exchange type

BasePublisher

Option	Type	Default	Description
exchange	string	required	Exchange name
exchangeType	ExchangeType	topic	Exchange type
confirm	boolean	true	Use publisher confirms
maxRetries	number	3	Max publish retries
initialRetryDelay	number	100	Initial retry delay (ms)
maxRetryDelay	number	5000	Max retry delay (ms)
useCircuitBreaker	boolean	true	Enable circuit breaker

Health & Metrics

import { HealthService, ConnectionStatus } from 'amqp-resilient';

// Get overall health status
const status = HealthService.getOverallStatus();
// 'healthy' | 'degraded' | 'dead' | 'not_configured'

// Get specific connection status
const connStatus = HealthService.getStatus('my-connection');
// ConnectionStatus.CONNECTED | CONNECTING | DISCONNECTED | RECONNECTING | CLOSED

// Get all connection statuses
const allStatuses = HealthService.getAllStatuses();
// { 'my-connection': 'connected', 'other-connection': 'reconnecting' }

// Get connection stats
const stats = connection.getStats();
// {
//   connected: true,
//   reconnectAttempts: 0,
//   channelCount: 2,
//   lastConnectedAt: Date,
// }

Events

connection.on('connected', () => {
  console.log('AMQP connected');
});

connection.on('disconnected', () => {
  console.log('AMQP disconnected');
});

connection.on('reconnecting', (attempt) => {
  console.log(`Reconnecting attempt ${attempt}`);
});

connection.on('error', (error) => {
  console.error('Connection error:', error);
});

Circuit Breaker

Standalone circuit breaker that can be used independently:

import { CircuitBreaker, CircuitBreakerOpenError } from 'amqp-resilient';

const breaker = new CircuitBreaker({
  name: 'external-api',
  failureThreshold: 5,    // Open after 5 failures
  resetTimeout: 30000,    // Try half-open after 30s
  successThreshold: 3,    // Close after 3 successes in half-open
});

try {
  const result = await breaker.execute(async () => {
    return await externalApiCall();
  });
} catch (error) {
  if (error instanceof CircuitBreakerOpenError) {
    console.log(`Circuit open, retry in ${error.remainingResetTime}ms`);
  }
}

// Get circuit state
const state = breaker.getState();
// 'CLOSED' | 'OPEN' | 'HALF_OPEN'

Logger Interface

Any logger that implements this interface works:

interface AmqpLogger {
  info(obj: object, msg?: string): void;
  warn(obj: object, msg?: string): void;
  error(obj: object, msg?: string): void;
  debug(obj: object, msg?: string): void;
}

// Examples: pino, winston, bunyan, console

Dead Letter Queue (DLQ)

Messages that fail after max retries are automatically sent to a DLQ:

// Queue: orders
// DLQ: orders.dlq (auto-created)

// Message headers in DLQ:
// x-death-reason: 'max-retries-exceeded'
// x-death-count: 3
// x-original-routing-key: 'order.created'
// x-first-death-queue: 'orders'

Requirements

• Node.js >= 18.0.0
• amqplib >= 0.10.0

License

MIT berkeerdo