@message-in-the-middle/pipe-adapters-rabbitmq

⚠️ Work in Progress Is this library production-ready? No. Is this library safe? No. When will it be ready? Soon™ (maybe tomorrow, maybe never). Why is it public? Experiment

message-in-the-middle is to Express.js what your message queue processing is to HTTP request processing. Just as Express provides a middleware pattern for HTTP requests, this library provides a middleware pattern for processing queue messages.

Why This Exists

Processing queue messages usually means copy-pasting the same boilerplate: parse JSON, validate, log, retry, deduplicate, route to handlers. This library lets you compose that logic as middlewares.

RabbitMQ (AMQP) adapter for message pipeline orchestration

Queue adapter implementation for RabbitMQ, enabling pipeline orchestration with RabbitMQ queues and exchanges.

Installation

npm install @message-in-the-middle/pipe-core @message-in-the-middle/pipe-adapters-rabbitmq amqplib
npm install --save-dev @types/amqplib

Quick Start

import { MessagePipelineOrchestrator } from '@message-in-the-middle/pipe-core';
import { RabbitMQPipeAdapter } from '@message-in-the-middle/pipe-adapters-rabbitmq';

const pipeline = new MessagePipelineOrchestrator()
  .source('orders', new RabbitMQPipeAdapter({
    connection: 'amqp://localhost',
    queue: 'orders',
    prefetch: 10,
  }))
  .pipe('process', processOrder)
  .destination('processed', new RabbitMQPipeAdapter({
    connection: 'amqp://localhost',
    queue: 'processed-orders',
    exchange: 'orders-exchange',
    routingKey: 'order.processed',
  }))
  .start();

Features

• ✅ Multiple exchange types (direct, topic, fanout, headers)
• ✅ Prefetch (QoS) control
• ✅ Queue and exchange assertion
• ✅ Message priority and TTL support
• ✅ Dead letter exchange configuration
• ✅ Connection sharing across adapters
• ✅ Manual acknowledgment support

API

RabbitMQPipeAdapter Options

interface RabbitMQPipeAdapterOptions {
  connection: Connection | string | Options.Connect; // Required
  queue: string;                          // Required: Queue name
  name?: string;                          // Optional: Custom adapter name
  exchange?: string;                      // Optional: Exchange name (default: '')
  routingKey?: string;                    // Optional: Routing key (default: queue name)
  queueOptions?: Options.AssertQueue;     // Optional: Queue assertion options
  exchangeOptions?: Options.AssertExchange; // Optional: Exchange assertion options
  prefetch?: number;                      // Default: 10
  noAck?: boolean;                        // Default: false (manual ack)
  assertQueue?: boolean;                  // Default: true
  assertExchange?: boolean;               // Default: false
  logger?: Logger;                        // Optional: Logger instance
}

Example with Exchange

const adapter = new RabbitMQPipeAdapter({
  connection: 'amqp://localhost',
  queue: 'orders',
  exchange: 'orders-exchange',
  routingKey: 'order.created',
  exchangeOptions: {
    type: 'topic',
    durable: true,
  },
  queueOptions: {
    durable: true,
    arguments: {
      'x-message-ttl': 86400000,          // 24 hours TTL
      'x-dead-letter-exchange': 'dlx',    // Dead letter exchange
    },
  },
  prefetch: 10,
  logger: console,
});

Advanced Features

Connection Sharing

Share a single connection across multiple adapters:

import * as amqp from 'amqplib';

const connection = await amqp.connect('amqp://localhost');

const sourceAdapter = new RabbitMQPipeAdapter({
  connection,  // Shared connection
  queue: 'source-queue',
});

const destAdapter = new RabbitMQPipeAdapter({
  connection,  // Same connection
  queue: 'dest-queue',
});

Message Priority

await adapter.send(message, {
  custom: {
    priority: 10,  // 0-10, higher = more priority
  },
});

Message TTL

await adapter.send(message, {
  custom: {
    expiration: '60000',  // 60 seconds (string!)
  },
});

Examples

See examples/pipelines for complete examples:

• rabbitmq-to-rabbitmq-simple.ts - Basic RabbitMQ → RabbitMQ pipeline
• cross-queue-sqs-to-rabbitmq.ts - Cross-queue integration

Docker Setup

For local development and testing:

# docker-compose.yml
version: '3.8'
services:
  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - '5672:5672'
      - '15672:15672'
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest

docker-compose up -d

License

MIT