@roasmax/rabbitmq

A comprehensive RabbitMQ abstraction framework for TypeScript/Node.js with enterprise-grade features, type safety, and high performance.

✨ Features

• 🚀 High Performance - Optimized for throughput and low latency
• 🔒 Type Safe - Full TypeScript support with runtime validation
• 🏗️ Enterprise Ready - Connection pooling, middleware, monitoring
• 🎯 Pattern Based - Work Queue, Pub/Sub, Router, RPC patterns
• 🔧 Extensible - Plugin architecture with middleware support
• 📊 Observable - Built-in metrics, logging, and tracing
• 🛡️ Reliable - Comprehensive error handling and retry mechanisms
• 📚 Well Documented - Extensive documentation and examples

🚀 Quick Start

Installation

npm install @roasmax/rabbitmq
# or
yarn add @roasmax/rabbitmq
# or
pnpm add @roasmax/rabbitmq

Basic Usage

import { RabbitMQClient, createFunctionHandler } from '@roasmax/rabbitmq';

// Create and initialize client
const client = RabbitMQClient.create({
  host: 'localhost',
  port: 5672,
  username: 'guest',
  password: 'guest'
});

await client.initialize();

// Work Queue Pattern
const workQueue = client.createWorkQueue('task_queue');
await workQueue.setupQueue();

// Create task handler
const taskHandler = createFunctionHandler(async (taskData) => {
  console.log('Processing task:', taskData);
  await new Promise(resolve => setTimeout(resolve, 1000)); // Simulate work
  return { processed: true };
});

// Start workers
await workQueue.startWorkers(taskHandler, 3);

// Send tasks
await workQueue.sendTask({ type: 'email', to: '[email protected]' });
await workQueue.sendTask({ type: 'sms', to: '+1234567890' });

📋 Table of Contents

• Installation
• Quick Start
• Core Concepts
• Messaging Patterns
• Type Safety
• Middleware
• Enterprise Features
• Performance
• Configuration
• Examples
• API Documentation
• Contributing
• License

🏗️ Core Concepts

RabbitMQClient

The main client class provides a high-level interface to RabbitMQ:

// From configuration object
const client = RabbitMQClient.create({
  host: 'localhost',
  port: 5672,
  username: 'guest',
  password: 'guest'
});

// From environment variables
const client = RabbitMQClient.fromEnvironment();

// From configuration file
const client = RabbitMQClient.fromConfigFile('config.yaml');

Messages

Type-safe message handling with validation:

import { Message, createMessage } from '@roasmax/rabbitmq';

// Create message
const message = createMessage({ data: 'Hello World' })
  .withHeader('x-source', 'api')
  .withPriority(5)
  .build();

// Typed messages
interface UserEvent {
  userId: string;
  action: string;
  timestamp: string;
}

const typedMessage = Message.typed<UserEvent>({
  userId: '123',
  action: 'login',
  timestamp: new Date().toISOString()
});

🎯 Messaging Patterns

Work Queue

Distribute tasks among multiple workers:

const workQueue = client.createWorkQueue('heavy_tasks');
await workQueue.setupQueue();

// Send tasks
await workQueue.sendTask({
  type: 'image_processing',
  imageUrl: 'https://example.com/image.jpg'
});

// Process tasks
const processor = createFunctionHandler(async (task) => {
  return await processImage(task.imageUrl);
});

await workQueue.startWorkers(processor, 5); // 5 concurrent workers

Publish/Subscribe

Broadcast events to multiple subscribers:

const pubsub = client.createPubSub('events');
await pubsub.setupExchange();

// Publisher
await pubsub.publish({
  type: 'user_registered',
  userId: 123,
  email: '[email protected]'
});

// Subscriber
const eventHandler = createFunctionHandler(async (event) => {
  console.log('Received event:', event);
});

await pubsub.subscribe(eventHandler, { queueName: 'analytics' });

Router

Route messages based on patterns:

const router = client.createRouter('user_events');
await router.setupExchange();

// Send to specific routes
await router.send('user.login', { userId: 123 });
await router.send('user.purchase', { userId: 123, amount: 99.99 });

// Subscribe to patterns
await router.subscribe('user.*', userHandler);
await router.subscribe('*.purchase', purchaseHandler);

RPC

Request/response communication:

const rpc = client.createRPC('calculator');

// Server
await rpc.setupServer({
  add: async (params) => params.a + params.b,
  multiply: async (params) => params.a * params.b
});

// Client
await rpc.setupClient();
const result = await rpc.call('add', { a: 5, b: 3 }); // Returns 8

🔒 Type Safety

Full TypeScript support with runtime validation using Zod:

import { createTypedHandler } from '@roasmax/rabbitmq';
import { z } from 'zod';

// Define schema
const UserEventSchema = z.object({
  userId: z.string(),
  eventType: z.enum(['login', 'logout', 'purchase']),
  timestamp: z.string(),
  metadata: z.record(z.unknown()).optional()
});

type UserEvent = z.infer<typeof UserEventSchema>;

// Create typed handler
const userEventHandler = createTypedHandler(
  UserEventSchema,
  async (event: UserEvent) => {
    // TypeScript knows the exact type here
    console.log(`User ${event.userId} performed ${event.eventType}`);

    if (event.eventType === 'purchase') {
      await processPurchase(event);
    }

    return { processed: true };
  }
);

🔧 Middleware

Extensible middleware system for cross-cutting concerns:

import {
  createLoggingMiddleware,
  createMetricsMiddleware,
  createStructuredLoggingMiddleware
} from '@roasmax/rabbitmq';

const pipeline = client.getMiddlewarePipeline();

// Logging middleware
pipeline.add(createLoggingMiddleware({
  logLevel: 'info',
  includeMessageBody: true,
  maxBodyLength: 1000
}));

// Metrics middleware
const metricsMiddleware = createMetricsMiddleware(undefined, 10000); // Report every 10s
pipeline.add(metricsMiddleware);

// Structured logging
pipeline.add(createStructuredLoggingMiddleware({
  sendLevel: 'debug',
  processLevel: 'info',
  errorLevel: 'error'
}));

// Get metrics
const metrics = metricsMiddleware.getDetailedMetrics();
console.log(`Success rate: ${metrics.rates.successRate * 100}%`);

Custom Middleware

Create your own middleware:

import { Middleware, MiddlewareContext } from '@roasmax/rabbitmq';

class AuthMiddleware implements Middleware {
  async beforeProcess(context: MiddlewareContext): Promise<MiddlewareContext> {
    const token = context.message.getHeader('authorization');
    if (!token) {
      throw new Error('Missing authorization header');
    }

    const user = await validateToken(token);
    context.metadata.user = user;

    return context;
  }
}

pipeline.add(new AuthMiddleware());

🏢 Enterprise Features

Connection Pooling

Automatic connection management with pooling:

const client = RabbitMQClient.create({
  host: 'localhost',
  port: 5672,
  // Connection pool configuration
  poolSize: 20,
  acquireTimeout: 30000,
  idleTimeout: 300000
});

// Get connection statistics
const stats = client.getConnectionStats();
console.log(`Active connections: ${stats.busyConnections}/${stats.totalConnections}`);

Error Handling and Retries

Comprehensive error handling with configurable retry strategies:

import { RetryHandler, ConditionalHandler } from '@roasmax/rabbitmq';

// Retry handler with exponential backoff
const retryHandler = new RetryHandler(
  baseHandler,
  3, // max retries
  1000, // initial delay
  2 // backoff factor
);

// Conditional processing
const conditionalHandler = new ConditionalHandler()
  .when(
    (context) => context.message.getHeader('priority') === 'high',
    highPriorityHandler
  )
  .when(
    (context) => isBusinessHours(),
    businessHoursHandler
  )
  .otherwise(defaultHandler);

Performance Monitoring

Built-in performance testing and monitoring:

import { createPerformanceTester } from '@roasmax/rabbitmq';

const tester = createPerformanceTester(connectionPool);

// Run throughput test
const throughputResult = await tester.runThroughputTest(
  'test_queue',
  10000, // message count
  10,    // concurrent producers
  1024   // message size
);

console.log(`Throughput: ${throughputResult.throughput} messages/second`);

// Run latency test
const latencyResult = await tester.runLatencyTest('test_queue', 1000);
console.log(`P95 latency: ${latencyResult.latencyStats.p95}ms`);

// Run comprehensive test suite
const results = await tester.runTestSuite('test_queue');

⚙️ Configuration

Environment Variables

Configure using environment variables:

export RABBITMQ_HOST=localhost
export RABBITMQ_PORT=5672
export RABBITMQ_USERNAME=guest
export RABBITMQ_PASSWORD=guest
export RABBITMQ_VIRTUAL_HOST=/
export RABBITMQ_HEARTBEAT=600
export RABBITMQ_CONNECTION_TIMEOUT=30000
export RABBITMQ_SSL_ENABLED=false
export RABBITMQ_MAX_RETRIES=3
export RABBITMQ_RETRY_DELAY=1000

Configuration File

Create a rabbitmq.yaml configuration file:

host: localhost
port: 5672
username: guest
password: guest
virtualHost: /
heartbeat: 600
connectionTimeout: 30000
sslEnabled: false
sslOptions: {}
maxRetries: 3
retryDelay: 1000

Load configuration:

const client = RabbitMQClient.fromConfigFile('rabbitmq.yaml');

Programmatic Configuration

const client = RabbitMQClient.create({
  host: 'rabbitmq.example.com',
  port: 5672,
  username: 'myapp',
  password: 'secret',
  virtualHost: '/production',
  heartbeat: 300,
  connectionTimeout: 10000,
  sslEnabled: true,
  sslOptions: {
    rejectUnauthorized: true,
    ca: fs.readFileSync('ca-cert.pem'),
    cert: fs.readFileSync('client-cert.pem'),
    key: fs.readFileSync('client-key.pem')
  },
  maxRetries: 5,
  retryDelay: 2000
});

📊 Performance

Benchmarks

Performance characteristics on standard hardware:

• Throughput: 10,000+ messages/second
• Latency: P95 < 50ms, P99 < 100ms
• Concurrency: Supports hundreds of concurrent connections
• Reliability: 99.9%+ message delivery success rate

Optimization Tips

1. Use Connection Pooling: Reuse connections across operations
2. Batch Operations: Process messages in batches for high throughput
3. Async Processing: Use async/await for non-blocking operations
4. Proper Serialization: Choose appropriate serializers for your data
5. Monitor Performance: Use built-in metrics to identify bottlenecks

// High-performance configuration
const client = RabbitMQClient.create({
  host: 'localhost',
  port: 5672,
  username: 'guest',
  password: 'guest',
  heartbeat: 60,           // Shorter heartbeat for faster detection
  connectionTimeout: 5000, // Faster connection timeout
});

// Use concurrent consumers
const consumer = client.createConsumer(handler, true, 10); // 10 concurrent workers

// Use batch processing
const batchConsumer = client.createBatchConsumer(
  batchHandler,
  50,   // batch size
  5000  // timeout ms
);

📚 Examples

Basic Work Queue

import { RabbitMQClient, createFunctionHandler } from '@roasmax/rabbitmq';

async function basicWorkQueue() {
  const client = RabbitMQClient.create();
  await client.initialize();

  const workQueue = client.createWorkQueue('email_queue');
  await workQueue.setupQueue();

  // Producer
  await workQueue.sendTask({
    to: '[email protected]',
    subject: 'Welcome!',
    body: 'Welcome to our service!'
  });

  // Consumer
  const emailHandler = createFunctionHandler(async (emailData) => {
    await sendEmail(emailData);
    return { sent: true };
  });

  await workQueue.startWorkers(emailHandler, 3);
}

Enterprise Event Processing

import {
  RabbitMQClient,
  createTypedHandler,
  createLoggingMiddleware,
  createMetricsMiddleware
} from '@roasmax/rabbitmq';
import { z } from 'zod';

const OrderEventSchema = z.object({
  orderId: z.string(),
  customerId: z.string(),
  amount: z.number(),
  items: z.array(z.object({
    productId: z.string(),
    quantity: z.number(),
    price: z.number()
  }))
});

async function enterpriseEventProcessing() {
  const client = RabbitMQClient.create();
  await client.initialize();

  // Add middleware
  const pipeline = client.getMiddlewarePipeline();
  pipeline.add(createLoggingMiddleware({ logLevel: 'info' }));
  pipeline.add(createMetricsMiddleware());

  const router = client.createRouter('orders');
  await router.setupExchange();

  // Type-safe order processing
  const orderHandler = createTypedHandler(
    OrderEventSchema,
    async (order) => {
      console.log(`Processing order ${order.orderId} for ${order.customerId}`);

      // Process order logic
      await processPayment(order);
      await updateInventory(order.items);
      await sendConfirmation(order.customerId);

      return { processed: true, orderId: order.orderId };
    }
  );

  await router.subscribe('order.created', orderHandler);

  // Send order event
  await router.send('order.created', {
    orderId: 'order_123',
    customerId: 'customer_456',
    amount: 99.99,
    items: [
      { productId: 'prod_1', quantity: 2, price: 49.99 }
    ]
  });
}

Microservices Communication

// Service A - Order Service
const orderRPC = client.createRPC('order_service');
await orderRPC.setupServer({
  createOrder: async (orderData) => {
    const order = await createOrder(orderData);

    // Publish event
    const eventBus = client.createPubSub('events');
    await eventBus.publish({
      type: 'order_created',
      orderId: order.id,
      customerId: order.customerId
    });

    return order;
  },

  getOrder: async ({ orderId }) => {
    return await getOrderById(orderId);
  }
});

// Service B - Inventory Service
const inventoryRPC = client.createRPC('inventory_service');
await inventoryRPC.setupClient();

const eventBus = client.createPubSub('events');
const eventHandler = createFunctionHandler(async (event) => {
  if (event.type === 'order_created') {
    // Update inventory when order is created
    await inventoryRPC.call('reserveItems', {
      orderId: event.orderId,
      items: event.items
    });
  }
});

await eventBus.subscribe(eventHandler, { queueName: 'inventory_events' });

📖 API Documentation

For detailed API documentation, see:

• API Reference - Complete API documentation
• Getting Started Guide - Step-by-step tutorial
• Examples - Comprehensive examples
• TypeDoc Generated Docs - Auto-generated API docs

🛠️ Development

Prerequisites

• Node.js 16.x or higher
• pnpm 8.x or higher
• RabbitMQ server

Setup

# Clone the repository
git clone https://github.com/roasmax/rabbitmq.git
cd rabbitmq

# Install dependencies
pnpm install

# Build the project
pnpm build

# Run tests
pnpm test

# Run examples
pnpm example:basic
pnpm example:enterprise
pnpm example:performance

Running Tests

# Unit tests
pnpm test

# Integration tests (requires RabbitMQ)
pnpm test:integration

# Coverage report
pnpm test:coverage

# Watch mode
pnpm test:watch

Docker Setup

Run RabbitMQ with Docker for development:

docker run -d --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  -e RABBITMQ_DEFAULT_USER=guest \
  -e RABBITMQ_DEFAULT_PASS=guest \
  rabbitmq:3.12-management

🤝 Contributing

We welcome contributions! Please see our Contributing Guide for details.

Development Workflow

1. Fork the repository
2. Create a feature branch: git checkout -b feature/amazing-feature
3. Make your changes
4. Add tests for your changes
5. Run the test suite: pnpm test
6. Commit your changes: git commit -m 'Add amazing feature'
7. Push to the branch: git push origin feature/amazing-feature
8. Open a Pull Request

Code Style

• Use TypeScript strict mode
• Follow ESLint and Prettier configurations
• Write comprehensive tests
• Document public APIs with JSDoc
• Follow conventional commit messages

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

🙏 Acknowledgments

• amqplib - The underlying AMQP library
• RabbitMQ - The message broker
• TypeScript - For type safety
• Zod - For runtime type validation
• Pino - For high-performance logging

📞 Support

• 📚 Documentation
• 🐛 Issue Tracker
• 💬 Discussions
• 📧 Email Support

@roasmax/rabbitmq - Enterprise-grade RabbitMQ abstraction for TypeScript/Node.js

Made with ❤️ by roasmax