RabbitMQ Event Bus

A powerful and flexible Event Bus built on top of RabbitMQ, enabling reliable event-driven communication between multiple Node.js services. This package supports advanced features such as message persistence, retries, dead-letter queues, and idempotency.

Features

• Reliable Event Publishing and Consumption: Ensures message delivery with options for persistence and mandatory routing.
• Flexible Subscription: Supports multiple routing keys and handlers for a single queue.
• Dead Letter Queue (DLQ) Support: Captures failed messages for analysis or retries.
• Retry Mechanism: Automatically retries failed messages with configurable delays.
• Idempotency Handling: Avoids duplicate processing of the same message.
• Dynamic Configuration: Fully customizable message stores, retry logic, and connection options.

Installation

npm install @nodesandbox/event-bus

Usage

Initialization

Create an instance of the RabbitMQEventBus with the necessary configuration.

import { RabbitMQEventBus } from '@nodesandbox/event-bus';

const eventBus = new RabbitMQEventBus({
  connection: {
    url: 'amqp://localhost:5672',
    exchange: 'events',
    exchangeType: 'topic',
    deadLetterExchange: 'events.dlx',
    retryExchange: 'events.retry',
  },
  producer: {
    persistent: true,
    mandatory: true,
  },
  consumer: {
    prefetch: 10,
    autoAck: false,
  },
});

await eventBus.init();
console.log('RabbitMQ Event Bus initialized successfully.');

---

Publishing Events

Publish events to RabbitMQ using the publish method. Events follow the CloudEvents specification.

const event = {
  type: 'order.created',
  data: { orderId: '1234', userId: 'user001' },
  metadata: {
    id: 'event-id-1234',
    source: 'order-service',
    time: new Date().toISOString(),
  },
};

await eventBus.publish(event);
console.log('Event published:', event);

---

Subscribing to Events

Subscribe to events using the subscribe method and provide a handler function.

await eventBus.subscribe(
  ['order.created', 'order.updated'],
  async (event) => {
    console.log('Received event:', event);
    // Handle event
  },
  { queue: 'order-queue' } // Optional custom queue name
);

---

Unsubscribing

Unsubscribe a consumer using its consumer tag.

await eventBus.unsubscribe('consumer-tag');
console.log('Unsubscribed successfully.');

---

Closing the Event Bus

Gracefully close the Event Bus, including all active connections and channels.

await eventBus.close();
console.log('RabbitMQ Event Bus closed.');

---

Advanced Configuration

Dead Letter Queue (DLQ)

Capture failed messages using DLQ:

await eventBus.subscribe(
  ['events.dlx'],
  async (event) => {
    console.warn('Dead Letter Message:', event);
    // Handle DLQ message
  },
  { queue: 'dlq-queue' }
);

---

Retry Mechanism

Customize retry delays and maximum attempts:

const eventBus = new RabbitMQEventBus({
  connection: {
    url: 'amqp://localhost:5672',
    exchange: 'events',
  },
  producer: {
    persistent: true,
  },
  consumer: {
    prefetch: 10,
    autoAck: false,
  },
  retryOptions: {
    maxRetries: 5,
    retryDelays: [1000, 5000, 10000], // Retry after 1s, 5s, and 10s
  },
});

---

Idempotency Handling

The IdempotencyStore ensures that duplicate messages are not processed:

eventBus.useIdempotencyStore(new CustomIdempotencyStore());

class CustomIdempotencyStore {
  private processedMessages = new Map();

  async markAsProcessed(messageId, result) {
    this.processedMessages.set(messageId, { result });
  }

  async hasBeenProcessed(messageId) {
    return this.processedMessages.has(messageId);
  }
}

---

API Reference

RabbitMQEventBus(options)

Option	Description	Default
connection.url	The RabbitMQ connection URL.	amqp://localhost:5672
connection.exchange	Name of the exchange.	events
connection.exchangeType	Type of the exchange (topic, direct, fanout).	topic
producer.persistent	Ensure messages are persistent.	true
producer.mandatory	Return unroutable messages to the publisher.	true
consumer.prefetch	Number of messages to prefetch.	10
consumer.autoAck	Automatically acknowledge messages.	false

---

publish(event, options?)

Publishes an event to RabbitMQ.

• event: The CloudEvent to be published.
• options: Additional AMQP options (e.g., headers, expiration).

---

subscribe(eventTypes, handler, options?)

Subscribes to one or more event types.

• eventTypes: An array of event types or a single event type.
• handler: A function to handle incoming events.
• options.queue: The name of the queue.

---

unsubscribe(consumerTag)

Unsubscribes a consumer by its tag.

---

close()

Closes the connection and all active channels.

---

Best Practices

• Use Persistent Messages: Ensure critical messages are not lost by enabling persistence.
• Implement DLQs: Capture and analyze failed messages for debugging or retrying.
• Use Idempotency: Prevent duplicate processing of messages.
• Monitor RabbitMQ: Leverage RabbitMQ's monitoring tools to track queue states and message flow.

---

Sample Project: A Practical Example

To see the package in action, we created a sample project simulating an e-commerce system with multiple microservices communicating via this Event Bus.

Overview of the Sample Project

• Order Service:

  • Publishes events related to order creation (order.created) and updates.
  • Interacts with Inventory and Payment Services to coordinate order processing.

• Inventory Service:

  • Subscribes to stock-related events like stock.check and stock.reserved.
  • Publishes responses (stock.check.response) and manages inventory updates.

• Payment Service:

  • Handles payment-related events such as payment.initiated and publishes results (payment.succeeded or payment.failed).

• Notification Service:

  • Subscribes to multiple events (order.created, order.completed, payment.succeeded, etc.) to send notifications.

Each service is a separate Node.js server using this package for communication. The project demonstrates reliable event flow, retries, and error handling.

How to Run the Sample Project

1. Clone the sample project repository:

   git clone https://github.com/nodesandbox/event-bus-sample.git
   cd event-bus-sample

2. Install dependencies:

   npm install

3. Start RabbitMQ: Ensure RabbitMQ is running, either locally or via Docker:

   docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:management

4. Start the services: Launch each service in a separate terminal:

   # Order Service
   cd order-service
   npm start
   
   # Inventory Service
   cd inventory-service
   npm start
   
   # Payment Service
   cd payment-service
   npm start
   
   # Notification Service
   cd notification-service
   npm start

5. Test the system: Use Postman, Curl, or any HTTP client to simulate creating an order:

   curl -X POST http://localhost:3001/orders \
   -H "Content-Type: application/json" \
   -d '{
       "userId": "user123",
       "items": [
           { "productId": "PROD1", "quantity": 2, "price": 19.99 },
           { "productId": "PROD2", "quantity": 1, "price": 9.99 }
       ]
   }'

   Check the logs of all services to observe event publication and handling.

6. Monitor with RabbitMQ Management UI: Access the RabbitMQ management interface at http://localhost:15672/. Use guest/guest as login credentials to view exchanges, queues, and messages.

Repository

Find the full source code for the sample project here:

https://github.com/nodesandbox/event-bus-sample

---

Contributing

We welcome contributions! Please follow these steps:

1. Fork the repository.
2. Create a new branch.
3. Submit a pull request with your changes.

---

License

This package is licensed under the MIT License. See LICENSE for more details.