@milan-dani/message-broker

Lightweight and reliable NATS/JetStream-based message broker for microservices (Node.js CommonJS)

This package provides a simple abstraction over NATS (and JetStream), allowing your services to easily publish and subscribe to events in an event-driven architecture — without manually handling low-level NATS details.

🚀 Features

• ⚡ Simple setup using initBroker()
• 📡 Easy publish/subscribe interface (emit / subscribe)
• 🧩 Built-in connection handling & automatic reconnection
• 💾 Outbox pattern and event consistency helpers (optional)
• 🧠 Suitable for distributed microservices

📦 Installation

npm install @milan-dani/message-broker

or

yarn add @milan-dani/message-broker

🧠 Quick Usage Example

Initialize the broker in your service

const { initBroker } = require("@milan-dani/message-broker");

const SERVICE_NAME = "orders-service";
const JS_STREAM = "ORDERS_STREAM";

async function start() {
  // Initialize broker connection
  const broker = await initBroker({
    serviceName: SERVICE_NAME,
    stream: JS_STREAM,
  });

  // small delay helps stabilize connection
  await new Promise((r) => setTimeout(r, 500));

  // Register your subscriptions
  await subscriptionHandler(broker);

  // Example of emitting an event
  await broker.emit("order.paid", {
    orderId: "ORD-12345",
    userId: "USR-789",
    transactionId: "TXN-555",
    amount: 99.99,
    items: [{ sku: "ITEM-001", qty: 2 }],
  });
}

start().catch(console.error);

Handle subscriptions

async function subscriptionHandler(broker) {
  // Listen for payment events
  broker.subscribe("order.paid", async (data, m) => {
    console.log("💰 Payment success for order:", data.orderId);

    // Update order status, trigger notifications, etc.
    // Example:
    // await updateOrderStatus(data.orderId, "PAID");
  });

  // You can subscribe to multiple subjects
  broker.subscribe("order.cancelled", async (data) => {
    console.log("❌ Order cancelled:", data.orderId);
  });
}

🧩 API Reference

initBroker(options)

Initialize a NATS connection and return a broker instance.

| Name | Type | Required | Description | | --------------------- | -------- | -------- | ------------------------------------------------------------ | | options.serviceName | string | ✅ | Unique name of your service (used for durable subscriptions) | | options.stream | string | ✅ | JetStream stream name to use | | options.url | string | ❌ | NATS server URL (default: nats://localhost:4222) |

Returns: broker instance with helper methods.

broker.subscribe(subject, handler)

Subscribe to an event.

broker.subscribe("order.created", async (data, message) => {
  console.log("Received:", data);
});

| Name | Type | Description | | --------- | ------------------------- | --------------------------- | | subject | string | NATS subject to listen to | | handler | function(data, message) | Called when message arrives |

broker.emit(subject, payload)

Publish an event.

await broker.emit("order.paid", { orderId: "ORD-123" });

broker.close()

Gracefully close the connection.

await broker.close();

🧱 Folder Structure

@/milan-dani/message-broker/
│
├── index.js
├── monitor.js
├── eventEmitter.js
├── natsClient.js
├── outbox.js
├── subscriber.js
└── package.json

🚀 Publishing to npm

To publish your package publicly to npm:

npm login
npm publish --access public

Make sure your package.json includes these fields:

{
  "name": "@milan-dani/message-broker",
  "version": "0.1.0",
  "main": "index.js",
  "license": "MIT",
  "description": "Lightweight NATS/JetStream message broker for Node.js microservices"
}

🤝 Contributing

1. Fork the repo
2. Create a new feature branch
3. Commit changes & open a PR 🎉

🧾 License

MIT License — see LICENSE

💬 Author

Milan Dani @milan-dani