Messaging with RabbitMQ: Asynchronous Communication between Microservices

// src/config/rabbitmq.config.tsexport interface RabbitMQConfig {  host: string;  port: number;  username: string;  password: string;  vhost: string;  heartbeat: number;  prefetch: number;  reconnectDelay: number;  maxReconnectAttempts: number;} export const rabbitmqConfig: RabbitMQConfig = {  host: process.env.RABBITMQ_HOST || 'localhost',  port: parseInt(process.env.RABBITMQ_PORT || '5672'),  username: process.env.RABBITMQ_USER || 'guest',  password: process.env.RABBITMQ_PASS || 'guest',  vhost: process.env.RABBITMQ_VHOST || '/',  heartbeat: 60,  prefetch: 10,  reconnectDelay: 5000,  maxReconnectAttempts: 10,}; export const exchanges: Record<string, ExchangeConfig> = {  orders: {    name: 'orders.exchange',    type: 'topic',    durable: true,    autoDelete: false,  },  notifications: {    name: 'notifications.exchange',    type: 'fanout',    durable: true,    autoDelete: false,  },  deadLetter: {    name: 'dlx.exchange',    type: 'direct',    durable: true,    autoDelete: false,  },}; export const queues: Record<string, QueueConfig> = {  orderCreated: {    name: 'orders.created.queue',    durable: true,    exclusive: false,    autoDelete: false,    deadLetterExchange: 'dlx.exchange',    deadLetterRoutingKey: 'orders.created.dlq',    messageTtl: 86400000, // 24 hours  },  orderPaid: {    name: 'orders.paid.queue',    durable: true,    exclusive: false,    autoDelete: false,    deadLetterExchange: 'dlx.exchange',    deadLetterRoutingKey: 'orders.paid.dlq',  },  notifications: {    name: 'notifications.queue',    durable: true,    exclusive: false,    autoDelete: false,    maxLength: 10000,  },  deadLetter: {    name: 'dead-letter.queue',    durable: true,    exclusive: false,    autoDelete: false,  },};

// src/infrastructure/messaging/RabbitMQConnection.tsimport amqp, { Connection, Channel, ConfirmChannel } from 'amqplib';import { EventEmitter } from 'events'; export class RabbitMQConnection extends EventEmitter {  private connection: Connection | null = null;  private channel: ConfirmChannel | null = null;  private reconnectAttempts = 0;  private isConnecting = false;   constructor(    private readonly config: RabbitMQConfig,    private readonly logger: Logger  ) {    super();  }   async connect(): Promise<void> {    if (this.isConnecting) return;    this.isConnecting = true;     try {      const url = this.buildConnectionUrl();      this.connection = await amqp.connect(url, {        heartbeat: this.config.heartbeat,      });       this.connection.on('error', (err) => {        this.logger.error('RabbitMQ connection error', { error: err.message });        this.emit('error', err);      });       this.connection.on('close', () => {        this.logger.warn('RabbitMQ connection closed');        this.emit('disconnected');        this.scheduleReconnect();      });       this.channel = await this.connection.createConfirmChannel();      await this.channel.prefetch(this.config.prefetch);       this.reconnectAttempts = 0;      this.isConnecting = false;      this.emit('connected');      this.logger.info('Connected to RabbitMQ');    } catch (error) {      this.isConnecting = false;      this.scheduleReconnect();      throw error;    }  }   private scheduleReconnect(): void {    if (this.reconnectAttempts >= this.config.maxReconnectAttempts) {      this.logger.error('Max reconnect attempts reached');      this.emit('maxReconnectAttempts');      return;    }     this.reconnectAttempts++;    const delay = this.config.reconnectDelay * this.reconnectAttempts;     this.logger.info(`Scheduling reconnect attempt ${this.reconnectAttempts}`, {      delayMs: delay,    });     setTimeout(() => this.connect(), delay);  }   getChannel(): ConfirmChannel {    if (!this.channel) throw new Error('Channel not initialized');    return this.channel;  }   async close(): Promise<void> {    if (this.channel) await this.channel.close();    if (this.connection) await this.connection.close();    this.logger.info('RabbitMQ connection closed gracefully');  }   isConnected(): boolean {    return this.connection !== null && this.channel !== null;  }}

// src/domain/events/DomainEvent.tsexport interface DomainEvent<T = unknown> {  eventId: string;  eventType: string;  aggregateId: string;  aggregateType: string;  timestamp: Date;  version: number;  payload: T;  metadata: EventMetadata;} export interface EventMetadata {  correlationId: string;  causationId?: string;  userId?: string;  traceId?: string;  spanId?: string;} // src/domain/events/OrderCreatedEvent.tsexport interface OrderCreatedPayload {  orderId: string;  customerId: string;  items: Array<{    productId: string;    quantity: number;    price: number;  }>;  totalAmount: number;  currency: string;  shippingAddress: {    street: string;    city: string;    state: string;    zipCode: string;    country: string;  };} export class OrderCreatedEvent extends BaseDomainEvent<OrderCreatedPayload> {  static readonly EVENT_TYPE = 'order.created';   constructor(payload: OrderCreatedPayload, metadata: EventMetadata) {    super(      OrderCreatedEvent.EVENT_TYPE,      payload.orderId,      'Order',      payload,      metadata    );  }}

// src/infrastructure/messaging/RabbitMQConsumer.tsexport class RabbitMQConsumer {  private channel: ConfirmChannel | null = null;  private handlers: Map<string, IMessageHandler> = new Map();   constructor(    private readonly connection: RabbitMQConnection,    private readonly serializer: MessageSerializer,    private readonly logger: Logger,    private readonly options: ConsumerOptions  ) {    this.connection.on('connected', () => {      this.channel = this.connection.getChannel();    });  }   registerHandler(eventType: string, handler: IMessageHandler): void {    this.handlers.set(eventType, handler);    this.logger.info(`Handler registered for event type: ${eventType}`);  }   async start(): Promise<void> {    if (!this.channel) throw new Error('Channel not available');     await this.channel.consume(      this.options.queue,      (msg) => this.handleMessage(msg),      { noAck: this.options.noAck ?? false }    );     this.logger.info(`Consumer started for queue: ${this.options.queue}`);  }   private async handleMessage(msg: ConsumeMessage | null): Promise<void> {    if (!msg || !this.channel) return;     const startTime = Date.now();    const eventType = msg.properties.type;     try {      const event = this.serializer.deserialize<DomainEvent>(msg.content);      const handler = this.handlers.get(eventType);       if (!handler) {        this.logger.warn(`No handler found for event type: ${eventType}`);        this.channel.ack(msg);        return;      }       await handler.handle(event);      this.channel.ack(msg);       this.logger.info('Message processed successfully', {        eventType,        durationMs: Date.now() - startTime,      });    } catch (error) {      await this.handleError(msg, error as Error);    }  }   private async handleError(msg: ConsumeMessage, error: Error): Promise<void> {    const retryCount = this.getRetryCount(msg);    const maxRetries = this.options.maxRetries ?? 3;     if (error instanceof RetryableError && retryCount < maxRetries) {      // Will be requeued with delay via dead letter exchange      this.channel!.nack(msg, false, false);    } else {      // Send to dead letter queue permanently      this.channel!.nack(msg, false, false);      this.logger.warn('Message sent to dead letter queue', {        reason: retryCount >= maxRetries ? 'max retries exceeded' : 'non-retryable error',      });    }  }}

// src/application/sagas/OrderSaga.tsinterface SagaStep<T> {  name: string;  execute: (data: T) => Promise<void>;  compensate: (data: T) => Promise<void>;} export class OrderSaga {  private steps: SagaStep<OrderSagaData>[] = [];  private executedSteps: SagaStep<OrderSagaData>[] = [];   async execute(data: OrderSagaData): Promise<void> {    this.logger.info('Starting order saga', { orderId: data.orderId });     for (const step of this.steps) {      try {        await step.execute(data);        this.executedSteps.push(step);      } catch (error) {        await this.compensate(data);        throw error;      }    }  }   private async compensate(data: OrderSagaData): Promise<void> {    // Compensate in reverse order    for (const step of this.executedSteps.reverse()) {      try {        await step.compensate(data);      } catch (error) {        this.logger.error(`Compensation failed for step: ${step.name}`);      }    }  }}

// src/infrastructure/outbox/OutboxProcessor.tsexport class OutboxProcessor {  private isRunning = false;   constructor(    private readonly repository: OutboxRepository,    private readonly publisher: IMessagePublisher,    private readonly intervalMs: number = 1000  ) {}   start(): void {    if (this.isRunning) return;    this.isRunning = true;    setInterval(() => this.processMessages(), this.intervalMs);  }   private async processMessages(): Promise<void> {    const messages = await this.repository.getUnprocessed(100);     for (const message of messages) {      try {        const event = JSON.parse(message.payload);        await this.publisher.publish(          `${message.aggregateType.toLowerCase()}.exchange`,          message.eventType,          event        );        await this.repository.markAsProcessed(message.id);      } catch (error) {        await this.repository.incrementRetry(message.id);      }    }  }}

// src/infrastructure/metrics/MessagingMetrics.tsexport class MessagingMetrics {  private readonly messagesPublished: Counter;  private readonly messagesConsumed: Counter;  private readonly messagesFailed: Counter;  private readonly messageProcessingDuration: Histogram;   recordPublish(exchange: string, routingKey: string, eventType: string): void {    this.messagesPublished.labels(exchange, routingKey, eventType).inc();  }   recordConsume(queue: string, eventType: string, status: 'success' | 'failed'): void {    this.messagesConsumed.labels(queue, eventType, status).inc();  }   recordProcessingDuration(queue: string, eventType: string, durationMs: number): void {    this.messageProcessingDuration.labels(queue, eventType).observe(durationMs / 1000);  }}