message-brokers

Message-Brokers is a package designed for managing RabbitMQ connections and operations. It simplifies the setup of queues and exchanges based on provided configurations and supports publishing messages to RabbitMQ with customizable routing keys and priorities. The package also integrates caching for better message management and retry mechanisms in case of failures.

Installation

Using npm

npm i message-brokers

Using yarn

yarn add message-brokers

Initialization

Set the following environment variables:

RABBITMQ_URL=amqp://admin:pwd@localhost:5672

(Optional)
ACTIVE_MESSAGE_BROKER=rabbitmq
# Defines the active message broker used by the system.
# Supported values: rabbitmq | redis (future support possible)

REDIS_PORT=6260
# Port used for Redis connection (used for caching / broker fallback mechanisms)

CACHING_ENABLED=true
# Enables/disables caching layer for message publishing and broker operations


RABBITMQ_DEPLOYMENT_ENV=stage
# Logical deployment identifier for shared RabbitMQ infrastructure.
# When set, it is appended to ALL RabbitMQ entities:
# - exchanges
# - queues
# - routing keys
#
# Example:
# base.queue → base.queue.stage
# base-key → base-key.stage
#
# Use this when multiple environments (stage/dev/prod) share the same RabbitMQ cluster.


DATACLOAK_DISABLE=true
# Disables Datacloak-related queues for non-production environments (e.g. stage/demo).
# When enabled, Datacloak queues will be skipped and removed if they exist.
# Useful for reducing unnecessary processing in shared or test deployments.


RABBITMQ_ISOLATED_NAMESPACE=prod-2
# Applies an isolation namespace ONLY to selected queues and routing keys.
#
# This is used together with an isolation prefix list (e.g. rabbitmqIsolatedQueues).
# If a queue/routing key matches the isolation rule, this namespace is appended:
#
# Example:
# messaging.queue → messaging.queue.prod-2
#
# Purpose:
# - Enables tenant-level or client-level isolation within the same broker
# - Prevents cross-communication between shared deployments
#
# Note:
# This does NOT apply globally — only to explicitly configured isolated queues/routing keys.

// ES5 Syntax
const MessageBrokers = require("message-brokers");
// ES6 Syntax
import MessageBrokers from "message-brokers";

const messageBrokers = new MessageBrokers();

Initialization with MongoDB for deadqueue

1. Mongoose initialization

const MessageBrokers = require("message-brokers");
const mongoose = require("mongoose");

const mongoConnectionHolder = { connection: null };

mongoose.connect("mongodb://localhost:27017/db");

const db = mongoose.connection;

db.on("connected", () => {
  console.log("✅ Mongoose connected!");
  mongoConnectionHolder.connection = db;
});

db.on("error", (err) => {
  console.error("❌ Mongoose connection error:", err);
});

const messageBrokers = new MessageBrokers(mongoConnectionHolder, "app_name");
await messageBrokers.init();

2. Native Mongodb driver initialization

const MessageBrokers = require("message-brokers");
const { MongoClient } = require("mongodb");

const mongoUrl = "mongodb://localhost:27017";
const dbName = "db";

const client = new MongoClient(mongoUrl, {
  useNewUrlParser: true,
  useUnifiedTopology: true,
});

const mongoConnectionHolder = { connection: null };

client
  .connect()
  .then(() => {
    console.log("✅ MongoDB connected!");
    mongoConnectionHolder.connection = client.db(dbName);
  })
  .catch((error) => console.error("❌ MongoDB connection failed:", error));

const messageBrokers = new MessageBrokers(mongoConnectionHolder, "app_name");
await messageBrokers.init();  

Usage

// Publish har
messageBrokers.publishHar(har);

// Publish har_meta
messageBrokers.publishHarMeta(har_meta);

// Publish lab_api_response
messageBrokers.publishLabApiResponse(lab_api_response);

// Publish purchase_details
messageBrokers.publishPurchaseDetails(purchase_details);

// Publish lab_location_data
messageBrokers.publishLabLocationData(lab_location_data);

// Publish input_object
messageBrokers.publishInputObject(input_object);

// Publish har_stub
messageBrokers.publishHarStub(har_stub);

// Publish raw_queue
messageBrokers.publishRawQueue(raw_queue);

// Publish notification_messages_logs
messageBrokers.publishNotificationMessagesLogs(notification_messages_logs);

// Publish amqp_lab_api_response
messageBrokers.publishAmqpLabApiResponse(amqp_lab_api_response);

// Publish billing_events
messageBrokers.publishBillingEvents(billing_events);