launchpad-rabbitmq

Helper package to interact with RabbitMQ. Handles connection management, consumer registration, publisher, exponential-backoff reconnect, and consumer health monitoring.

Setup

import { rabbitmq, consumer, publisher } from '@polymathnetwork/launchpad-rabbitmq';

rabbitmq.initialize({
  url: RABBITMQ_URL,
  retryDelay: RETRY_DELAY,
  maxRetries: MAX_RETRIES,

  // Reconnect policy (exponential backoff with jitter)
  reconnectDelay: 250,         // base delay for first reconnect attempt (ms)
  maxReconnectDelay: 30_000,   // hard cap on reconnect delay (ms)
  reconnectMultiplier: 2,      // backoff growth factor (default: 2)
  reconnectJitterRatio: 0.2,   // ±20% jitter on each delay (default: 0.2)

  // Health monitoring
  healthLogging: true,         // emit ERROR log when a consumer is lost (default: false)
});

await publisher.initialize();
await consumer.initialize();

// Start consumers — re-registered automatically after reconnect
await consumer.startConsumer(QUEUE_NAME, messageProcessor);

Publishing messages

import { publisher } from '@polymathnetwork/launchpad-rabbitmq';

const { ack } = await publisher.pushMessage(queueName, message);

Consumer health check

The consumer tracks whether its registered queues have active AMQP consumers. Use this in your /health endpoint so Kubernetes can restart the pod if consumers are lost (e.g. due to a TCP disconnect or heartbeat timeout):

import { consumer } from '@polymathnetwork/launchpad-rabbitmq';

app.get('/health', (_req, res) => {
  if (!consumer.isHealthy()) {
    return res.status(503).json({
      message: 'DEGRADED',
      reason: 'RabbitMQ consumers not active',
      missing: consumer.getMissingQueues(),
    });
  }
  res.json({ message: 'OK', uptime: process.uptime() });
});

isHealthy() returns true during startup (before any consumers are registered) and false only after at least one consumer has been registered and subsequently lost.

To toggle logging at runtime instead of via config:

consumer.enableLogging();   // start emitting ERROR logs on consumer loss
consumer.disableLogging();  // silence them again

HealthMonitor (standalone)

If you need a standalone health monitor instance (e.g. for a custom consumer setup not using the built-in consumer):

import { HealthMonitor } from '@polymathnetwork/launchpad-rabbitmq';

const monitor = new HealthMonitor({ logger: myLogger, logging: true });

// After channel.consume() resolves:
monitor.markActive(queueName);
channel.on('close', () => monitor.markInactive(queueName, 'channel closed'));
channel.on('error', (err) => monitor.markInactive(queueName, 'channel error', err.message));
channel.on('cancel', () => monitor.markInactive(queueName, 'consumer cancelled'));

monitor.isHealthy();       // boolean
monitor.getMissingQueues(); // string[]