npm package: @message-in-the-middle/pipe-adapters-rabbitmq
Description: RabbitMQ (AMQP) adapter for message pipeline orchestration
Author: Message in the Middle
Latest version: 0.1.2

# @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

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

## Quick Start

```typescript
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

```typescript
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

```typescript
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:

```typescript
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

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

### Message TTL

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

## Examples

See [examples/pipelines](../../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:

```yaml
# 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
```

```bash
docker-compose up -d
```

## License

MIT