# Introduction

## @nestjstools/messaging

```
npm install @nestjstools/messaging
or
yarn add @nestjstools/messaging
```

<figure><img src="/files/JEVspvJXM62VFYlJzjDr" alt=""><figcaption></figcaption></figure>

A NestJS library for managing asynchronous and synchronous messages (service bus) with support for buses, handlers, channels, and consumers. This library simplifies building scalable and decoupled applications by facilitating robust message handling pipelines while ensuring flexibility and reliability.

### Features

* **Message Buses**: Define multiple buses for commands, events, and queries to streamline message routing.
* **Handlers**: Easily register and manage handlers for processing messages.
* **Channels**: Support for in-memory channels and **easy extension** to create custom channel implementations tailored to your needs.
* **Consumers**: Run message consumers to process queued messages asynchronously, ensuring system reliability and fault tolerance.
* **Middleware Support**: Add custom middleware for message transformation such like validation, logging - do whatever you want.
* **Debug Mode**: Enable enhanced logging and debugging capabilities for development.
* **Extensibility**: Creating new channels is straightforward, allowing developers to expand and integrate with external systems or protocols effortlessly.
* **Concurrent Handler Execution**: Messages dispatched to multiple handlers are processed concurrently, improving performance and responsiveness across your system.

### Create an Event-driven app

<figure><img src="/files/yPowHhrqgTk3hXpHB6T6" alt=""><figcaption></figcaption></figure>

### Process logic in SYNC way in your handlers

<figure><img src="/files/q04bnCaU0FdHAuvKJkZn" alt=""><figcaption></figcaption></figure>

### Real-world app example

* <https://github.com/nestjstools/messaging-rabbitmq-example> - Messaging based on RabbitMQ + Redis
* <https://github.com/nestjstools/messaging-rabbitmq-example/tree/other-extensions> - All extensions

### Repositories

* <https://github.com/nestjstools/messaging>
* <https://github.com/nestjstools/messaging-bootstrap>


# Getting Started


# Installation

## 📦 Installation

To start using the `@nestjstools/messaging` library, you can install it using either `npm` or `yarn`, depending on your preference.

### Using npm

```bash
npm install @nestjstools/messaging
```

### Using yarn

```bash
yarn add @nestjstools/messaging
```

### 🧱 Requirements

* **Node.js** v22 or higher
* **NestJS 10+**


# Initialize Module

## 🧩 Defining the Messaging Module

To use `@nestjstools/messaging` in your application, you need to register the `MessagingModule` in your **root module** (usually `AppModule`). This will make the messaging functionality available across all other modules in your app.

### 🔧 Starter Setup

```ts
import { Module } from '@nestjs/common';
import { MessagingModule, InMemoryChannelConfig } from '@nestjstools/messaging';


@Module({
  imports: [
    MessagingModule.forRoot({
      buses: [
        {
          name: 'my-channel.bus',
          channels: ['my-channel'],
        },
      ],
      channels: [
        new InMemoryChannelConfig({
          name: 'my-channel',
        }),
      ],
      debug: true,
    }),
  ],
})
export class AppModule {}
```

### &#x20;Notes

* `MessagingModule.forRoot(...)` registers the messaging system globally, meaning you don't need to import it in feature modules.
* You must define at least one **bus** and one **channel**.
* `InMemoryChannelConfig` our app will works on in-memory message transport layer.
* Setting `debug: true` enables verbose logging to help with debugging message flow.

### &#x20;What’s Happening Here?

| Property     | Description                                                 |
| ------------ | ----------------------------------------------------------- |
| `buses`      | Defines logical buses for message routing.                  |
| `channels`   | Configures message channels (e.g., in-memory, Kafka, etc.). |
| `debug`      | Enables or disables debug logging.                          |
| `my-channel` | A simple in-memory channel used by the `message.bus`.       |

> ⚠️ **Important:** Make sure this is defined in your **AppModule** or the **main/root module** to ensure messaging is globally available.


# Message Handler

## 💬 Defining a Message & Message Handler

Messages are the core units of communication in your system. Each message can have a dedicated handler that processes it when received. This section explains how to define a **message** and implement a **handler**.

***

### 📨 Define Your Message

A message is a simple class representing the data being sent between services or components:

```ts
// send-message.ts
export class SendMessage {
  constructor(
    public readonly content: string,
  ) {}
}
```

This message can now be published on a bus and handled by one or more consumers.

***

### 🛠️ Define a Message Handler

Handlers process incoming messages. They must implement the `IMessageHandler<T>` interface.

```ts
import { Injectable } from '@nestjs/common';
import { MessageHandler, IMessageHandler, MessageResponse } from '@nestjstools/messaging';
import { SendMessage } from './send-message';

@Injectable()
@MessageHandler('your.message') // This should match the message route you publish to
export class SendMessageHandler implements IMessageHandler<SendMessage> {
  async handle(message: SendMessage): Promise<MessageResponse | void> {
    console.log(message.content);
    // Add your business logic here
  }
}
```

***

### 🔁 Multiple Routes

{% hint style="info" %}
You can associate a single handler with **multiple message routes**:

```ts
@MessageHandler('your.message', 'your.message2')
```

{% endhint %}

***

### Optional: Use `@DenormalizeMessage()` Decorator

If you want NestJS to automatically instantiate the incoming message as a proper `class` (rather than just receiving raw JSON), use the `@DenormalizeMessage()` decorator:

```ts
async handle(@DenormalizeMessage() message: SendMessage): Promise<MessageResponse | void> {
  // message is a real SendMessage instance
}
```


# Disaptch a message

## 📤 Dispatching a Message

Messages can be dispatched from anywhere in your application—such as services, controllers, scheduled jobs, or event listeners. This allows for clean, decoupled communication between parts of your system.

### 🚀 Example: Dispatch from an HTTP Controller

Here’s a simple example of dispatching a message when an HTTP request is made:

```ts
import { Controller, Get } from '@nestjs/common';
import { MessageBus, IMessageBus, RoutingMessage } from '@nestjstools/messaging';
import { SendMessage } from './test/send-message';

@Controller()
export class AppController {
  // You can inject any bus you've defined in your MessagingModule config
  constructor(@MessageBus('message.bus') private readonly messageBus: IMessageBus) {}

  @Get()
  async dispatchMessage(): Promise<string> {
    // Dispatching a SendMessage instance to the route 'your.message'
    await this.messageBus.dispatch(
      new RoutingMessage(new SendMessage('Message from HTTP request'), 'your.message'),
    );

    return 'Message dispatched successfully!';
  }
}
```

***

### 📦 Components Explained

| Component        | Purpose                                                          |
| ---------------- | ---------------------------------------------------------------- |
| `@MessageBus()`  | Injects a specific message bus by name.                          |
| `RoutingMessage` | Wraps the message and defines the route it should be sent to.    |
| `SendMessage`    | Your custom message class with data to transmit.                 |
| `dispatch()`     | Sends the message to all subscribed handlers matching the route. |

***

### 💡 Best Practices

* Use meaningful routing keys (e.g., `user.created`, `notification.send`).
* Define routing keys as the Enum

```
export enum EventMapper {
 UserCreated = 'my_app.event.user_created'
}
```

* Always wrap domain objects in proper message classes instead of sending raw payloads.


# Components


# Message Handlers

## 📦 Message Handlers

Message handlers define how your application reacts to specific messages dispatched on a bus. They encapsulate business logic triggered by inter-service or internal communication.

This page explains how to define, register, and work with message handlers using `@nestjstools/messaging`.

***

### 🛠️ Defining a Message Handler

A message handler is a class that implements the `IMessageHandler<T>` interface for a specific message type.

#### Example:

```ts
import { Injectable } from '@nestjs/common';
import { MessageHandler, IMessageHandler, MessageResponse } from '@nestjstools/messaging';
import { SendMessage } from './send-message';

@Injectable()
@MessageHandler('your.message')
export class SendMessageHandler implements IMessageHandler<SendMessage> {
  async handle(message: SendMessage): Promise<MessageResponse | void> {
    console.log(message.content);
    // Your business logic here
  }
}
```

***

### 🔁 Handling Multiple Routes

You can bind the same handler to **multiple routing keys**:

```ts
@MessageHandler('your.message', 'alternate.route')
```

This is useful if the same logic should run for different message sources.

***

### 🧠 Typed Message Input with `@DenormalizeMessage()`

By default, the handler receives the raw message data (usually as a plain JavaScript object). If you want to receive it as a properly instantiated class, use the `@DenormalizeMessage()` decorator:

```ts
async handle(@DenormalizeMessage() message: SendMessage): Promise<void> {
  // Now message is an instance of SendMessage with all class logic available
}
```

This is especially helpful when your message class contains methods, getters, or requires validation logic.

***

### 🧩 Registering Handlers

Handlers are automatically discovered by NestJS as long as they are included in the providers of a module:

```ts
@Module({
  providers: [SendMessageHandler],
})
export class MessagingFeatureModule {}
```

If you use a feature module, ensure that the module is imported by the root module and that `MessagingModule` is globally available or also imported.

***

### 📤 Triggering Handlers

Handlers respond to messages dispatched through a message bus with a matching route:

```ts
await messageBus.dispatch(
  new RoutingMessage(new SendMessage('Hello!'), 'your.message'),
);
```

If a handler is decorated with `@MessageHandler('your.message')`, it will receive this message.

***

### Returning a Response

Handlers can return a `MessageResponse` if needed:

```ts
import { MessageResponse } from '@nestjstools/messaging';

return new MessageResponse({ result: 'OK' });
```

You can optionally implement response-handling logic in the calling component.

***

### Summary

| Concept                 | Description                                                                         |
| ----------------------- | ----------------------------------------------------------------------------------- |
| `@MessageHandler()`     | Binds a handler to one or more routes                                               |
| `IMessageHandler<T>`    | Interface for handling a specific message type                                      |
| `@DenormalizeMessage()` | Automatically deserializes message into class instance                              |
| `MessageResponse`       | Optional way to return structured results from handlers, you can also return object |

***

{% hint style="info" %}
Note that handlers are visible across channels
{% endhint %}


# Normalizers

## 🧪 Normalizers

### What is a Normalizer?

A **Normalizer** is a component that transforms messages between different formats before sending and after receiving. It plays a crucial role in serialization and deserialization—ensuring that messages are properly encoded for transmission and correctly decoded when received.

This is especially helpful when integrating with systems that require specific data formats like **Protobuf**, **Base64**, or **custom JSON structures**.

***

### 🧩 Why Use a Normalizer?

You may want to use a custom normalizer to:

* Work with **binary formats** like Protobuf
* Encode messages in **Base64** or another custom format
* Encrypt/decrypt messages
* Support **custom serialization logic** not handled by default JSON

***

### ⚙️ Defining a Normalizer

To create a normalizer, implement the `MessageNormalizer` interface and decorate the class with `@MessagingNormalizer()`.

#### Example: Base64 Normalizer

```ts
import { Injectable } from '@nestjs/common';
import { MessagingNormalizer, MessageNormalizer } from '@nestjstools/messaging';
import { Buffer } from 'buffer';

@Injectable()
@MessagingNormalizer()
export class Base64Normalizer implements MessageNormalizer {
  denormalize(message: string | object, type: string): Promise<object> {
    if (typeof message === 'object') {
      throw new Error('Message must be a string!');
    }
    return Promise.resolve(
      JSON.parse(Buffer.from(message, 'base64').toString('utf-8')),
    );
  }

  normalize(message: object, type: string): Promise<string> {
    const jsonString = JSON.stringify(message);
    return Promise.resolve(
      Buffer.from(jsonString, 'utf-8').toString('base64'),
    );
  }
}
```

***

### 🔄 How It Works

| Phase           | Description                                                                                   |
| --------------- | --------------------------------------------------------------------------------------------- |
| **Normalize**   | Called before sending. Converts a message object into a transport-safe string (e.g., Base64). |
| **Denormalize** | Called after receiving. Converts the raw string back into a message object.                   |

***

### 🎯 Usage Per Channel

You can assign a normalizer **per channel** in your messaging configuration. This allows different channels to use different formats as needed:

```ts
new InMemoryChannelConfig({
  name: 'my-channel',
  normalizer: Base64Normalizer, // Use custom normalizer for this channel
}),
```

> 💡 Each channel can use a different normalizer depending on the protocol or service it interacts with.

***

### Summary

| Feature            | Description                                              |
| ------------------ | -------------------------------------------------------- |
| `normalize()`      | Transforms message before it’s sent                      |
| `denormalize()`    | Parses message after it’s received                       |
| Per-channel config | Apply specific format handling per communication channel |


# Exception Listeners

## ❗ Exception Listeners

In a distributed or asynchronous messaging system, things can go wrong: a handler might throw an error, a payload might be malformed, or some service dependency might fail. That’s where **Exception Listeners** come in.

The `@nestjstools/messaging` package provides a powerful way to **centrally manage and respond to unhandled exceptions** thrown during message processing.

{% hint style="warning" %}
This feature **does not** **work** for an **in-memory** channel
{% endhint %}

***

### 🔍 What is an Exception Listener?

An **Exception Listener** allows you to hook into the lifecycle of message processing and respond when any **uncaught exception** occurs during handling.

This is useful for:

* Logging errors
* Triggering alerting or monitoring workflows
* Performing recovery logic or fallback strategies
* Retrying logic (manual or delegated)

***

### 🛠️ How to Create One

To define an exception listener:

1. Implement the `ExceptionListener` interface
2. Decorate the class with `@MessagingExceptionListener()`

#### Example:

```ts
import { Injectable } from '@nestjs/common';
import {
  ExceptionListener,
  MessagingExceptionListener,
  ExceptionContext,
} from '@nestjstools/messaging';

@Injectable()
@MessagingExceptionListener()
export class CustomExceptionListener implements ExceptionListener {
  async onException(context: ExceptionContext): Promise<void> {
    console.log(`Exception caught during message handling:`, context.error);

    // Optional: Log to external services, trigger notifications, retry logic, etc.
  }
}
```

***

### 🧠 How It Works

* If an **exception** is thrown in any message handler or middleware during message execution:
  * It is **propagated up to the listener**
  * The `onException(context)` method is invoked with detailed metadata

#### The `ExceptionContext` includes:

| Field         | Description                                           |
| ------------- | ----------------------------------------------------- |
| `error`       | The exception that was thrown                         |
| `message`     | The message that caused the failure                   |
| `channelName` | The name of the channel where the error occurred      |
| `handler`     | (Optional) The handler class involved, if available   |
| `raw`         | Raw data passed to the channel before deserialization |

***

### Summary

| Feature                 | Description                                                  |
| ----------------------- | ------------------------------------------------------------ |
| 🧹 Centralized Handling | Avoid scattering try/catch blocks in every handler           |
| 🔁 Recovery Support     | Trigger fallback, retries, or compensating transactions      |
| 📈 Observability        | Report to monitoring tools like Sentry, Datadog, or Logstash |
| 🛠 Cleaner Code         | Keep message handlers focused on business logic              |


# Middlewares

## 🧵 Middlewares

In `@nestjstools/messaging` **Middlewares** are functions or classes that act on messages **before** they reach their respective handlers. They are ideal for adding **logging, validation, authentication, transformation, or metrics collection** to your messaging flow.

Each **channel** has its own middleware stack, and the middlewares are executed **in order**, just like an HTTP middleware pipeline.

***

### 🔍 What Does a Middleware Do?

A middleware has access to:

* The **incoming message** (`RoutingMessage`)
* The **execution context**
* The ability to:
  * Modify the message
  * Stop message processing
  * Forward the message to the next middleware or the final handler

***

### ✨ Use Cases

* Logging message flow
* Validating payloads
* Injecting metadata or tracing headers
* Authenticating messages
* Handling common errors or retries

***

### 🛠️ Creating a Middleware

Define a class that implements the `Middleware` interface and use the `@MessagingMiddleware()` decorator.

#### Example: Logging Middleware

```ts
import { Injectable } from '@nestjs/common';
import { Middleware, MessagingMiddleware, RoutingMessage, MiddlewareContext } from '@nestjstools/messaging';

@Injectable()
@MessagingMiddleware()
export class TestMiddleware implements Middleware {
  async process(message: RoutingMessage, context: MiddlewareContext): Promise<MiddlewareContext> {
    console.log('!!!! WORKS');
    
    // Continue processing the pipeline
    return await context.next().process(message, context);
  }
}
```

> 🔁 **Important:** Always call `context.next().process(...)` to continue down the middleware chain.

***

### 🔌 Attaching Middleware to a Channel

Middlewares are assigned per channel in your MessagingModule config:

```ts
import { MessagingModule, InMemoryChannelConfig, AmqpChannelConfig, ExchangeType } from '@nestjstools/messaging';
import { TestMiddleware } from './middlewares/test.middleware';

@Module({
  imports: [
    MessagingModule.forRoot({
      buses: [
        {
          name: 'message.bus',
          channels: ['my-channel'],
        },
      ],
      channels: [
        new InMemoryChannelConfig({
          name: 'my-channel',
          middlewares: [TestMiddleware],
        }),
      ],
      debug: true,
    }),
  ],
})
export class AppModule {}
```

***

### ⚙️ Middleware Pipeline

When a message is dispatched to a channel:

1. It passes through the **middleware stack**, in the order they’re defined.
2. Each middleware can:
   * Modify the message
   * Stop further processing
   * Pass the message on via `context.next().process(...)`
3. The final destination is the **message handler**.

***

### Benefits

| Feature                       | Description                                          |
| ----------------------------- | ---------------------------------------------------- |
| 🔄 **Reusable Logic**         | Write once, apply to multiple channels               |
| 🧼 **Separation of Concerns** | Keep logging/validation out of core handler logic    |
| 🔐 **Security & Validation**  | Centralize authentication and schema checks          |
| 🛠 **Customizable**           | Easily extend for metrics, tracing, throttling, etc. |


# Message Bus

## 📬 MessageBus

The `MessageBus` It is the **core interface** used to **dispatch messages** within your microservice architecture. It acts as the gateway between your application and the configured messaging channels, enabling seamless communication between services.

Whether you're calling from a controller, service, or any provider, the `MessageBus` gives you a powerful and type-safe way to route and send messages through your event-driven system.

***

### 🚀 What Is the MessageBus?

The `MessageBus` is injected via the `@MessageBus('bus.name')` decorator and gives you access to methods that allow you to:

* Send messages to specific handlers via routing
* Send raw or structured messages
* Target a specific **channel** (e.g., in-memory, AMQP)
* Trigger synchronous or asynchronous processing

***

### 🛠 Injecting the MessageBus

When configuring your messaging module, you define named buses. To use a bus, inject it using its name.

#### Example

```ts
import { Controller, Get } from '@nestjs/common';
import { MessageBus, IMessageBus, RoutingMessage } from '@nestjstools/messaging';
import { SendMessage } from './messages/send-message';

@Controller()
export class AppController {
  constructor(@MessageBus('message.bus') private readonly messageBus: IMessageBus) {}

  @Get()
  async dispatch(): Promise<string> {
    await this.messageBus.dispatch(
      new RoutingMessage(new SendMessage('Hello World!'), 'your.message'),
    );
    return 'Message dispatched!';
  }
}
```

***

### 🧩 RoutingMessage Structure

To route a message to the correct handler, use the `RoutingMessage` class:

```ts
new RoutingMessage(payload: object, route: string, messageOptions?: MessageOptions)
```

#### Example:

```ts
new RoutingMessage(
  new SendMessage('User created'),
  'user.created',
);
```

***

### 💡 Bus-to-Channel Mapping

In your configuration, each bus can be connected to multiple channels:

```ts
MessagingModule.forRoot({
  buses: [
    {
      name: 'message.bus',
      channels: ['my-channel'], // All messages from this bus go here
    },
  ],
  channels: [
    new InMemoryChannelConfig({
      name: 'my-channel',
    }),
  ],
})
```

> ✅ A message dispatched from a bus is passed to each of its configured channels.

***

### 🔄 One Bus, Many Channels

Each bus can dispatch to **multiple channels**, and you can even configure different channels (e.g., RabbitMQ + InMemory) to receive the same message for hybrid systems.

***

### 🔐 Benefits of MessageBus

| Feature      | Benefit                                                               |
| ------------ | --------------------------------------------------------------------- |
| ✅ Type-safe  | Dispatch class-based messages with strict typing                      |
| 🔌 Pluggable | Supports multiple channels per bus                                    |
| 📡 Scalable  | Easily add more channels or handlers without changing your core logic |
| 🧪 Testable  | Easily mock or simulate dispatches in testing environments            |


# Lifecycle Hooks

Messaging lifecycle hooks allow you to tap into different stages of message processing — from raw input to final handler execution.

They are useful for cross-cutting concerns like logging, tracing, validation, metrics, or custom transformations.

> **Important:** every processed object is treated as **immutable**.\
> Each stage works on a **new instance**, rather than mutating the previous one in place.

> **Important:** not every hook is guaranteed to run.\
> Some hooks may be skipped depending on the flow, for example when no handler is registered, processing stops early, or message handling fails before later stages are reached.

***

### Example hook

```typescript
import {
  HookMessage,
  LifecycleHook,
  MessagingLifecycleHook,
  MessagingLifecycleHookListener,
} from '@nestjstools/messaging';
import { Injectable, Logger } from '@nestjs/common';

@Injectable()
@MessagingLifecycleHook(LifecycleHook.AFTER_MESSAGE_HANDLER_EXECUTION)
export class AfterDenormalizeHook implements MessagingLifecycleHookListener {
  constructor(private readonly logger: Logger) {
  }

  hook(message: HookMessage): Promise<void> {
    this.logger.log(`💡 Here I can do some action ON HOOK | ${message.routingKey}`);
    return Promise.resolve();
  }
}

```

### Lifecycle Overview

```mermaid
flowchart TB
    A[Dispatch message]

    B(BEFORE_MESSAGE_NORMALIZATION)
    N[/Normalizer/]
    C(AFTER_MESSAGE_NORMALIZATION)
    DN[/Denormalizer/]
    D(AFTER_MESSAGE_DENORMALIZED)

    E(BEFORE_MESSAGE_HANDLER)
    M[/Middlewares/]
    H[/Handler/]

    X{THROW EXCEPTION}

    S(AFTER_MESSAGE_HANDLER_EXECUTION)
    ERR(ON_FAILED_MESSAGE_CONSUMER)

    A --> B --> N --> C --> DN --> D --> E --> M --> H --> X

    X -->|NO| S
    X -->|YES| ERR

    %% styles
    classDef hook fill:#000000,stroke:#ffffff,color:#ffffff,stroke-width:2px;
    classDef component fill:#5fa8ff,stroke:#1e88e5,color:#000000,stroke-width:2px;
    classDef middleware fill:#26c6da,stroke:#00838f,color:#000000,stroke-width:2px;
    classDef decision fill:#ff9800,stroke:#e65100,color:#ffffff,stroke-width:2px;

    class B,C,D,E,S,OK,ERR hook;
    class N,DN,H,A component;
    class M middleware;
    class X decision;
```

***

### Available Hooks

#### `BEFORE_MESSAGE_NORMALIZATION`

Executed before the raw incoming message is normalized into the internal messaging format.

Use this hook when you want to:

* inspect raw transport data,
* log the original payload,
* attach tracing or diagnostic metadata before normalization starts.

At this point, the message is still in its original external form.

***

#### `AFTER_MESSAGE_NORMALIZATION`

Executed after the raw message has been normalized into the internal messaging representation.

Use this hook when you want to:

* inspect the normalized structure,
* validate normalized metadata,
* enrich processing context based on the normalized message.

At this stage, the system has already transformed the transport payload into the unified internal format.

***

#### `AFTER_MESSAGE_DENORMALIZED`

Executed after the normalized message has been denormalized into the target application object.

Use this hook when you want to:

* inspect the final typed message object,
* perform additional validation on the resolved message instance,
* log the exact object that will be passed further into the pipeline.

This is usually the first point where you work with the final application-level message object instead of transport or normalized data.

***

#### `BEFORE_MESSAGE_HANDLER`

Executed right before the handler is invoked.

Use this hook when you want to:

* run last-step checks before handling,
* prepare contextual logging,
* inspect the message object that is about to be processed by the handler.

This hook is triggered after denormalization and before middleware / handler execution begins.

***

#### `AFTER_MESSAGE_HANDLER_EXECUTION`

Executed after the handler finishes successfully.

Use this hook when you want to:

* log successful processing,
* measure execution results,
* trigger post-processing actions after handler completion.

This hook is only executed when the handler completes without throwing an exception.

***

#### `ON_CONSUMER_HANDLED_MESSAGE`

Executed when a consumer has successfully handled the message.

Use this hook when you want to:

* emit success metrics,
* store processing audit entries,
* mark the message as fully processed.

This is a success-only hook and is typically one of the last lifecycle points in the handling flow.

***

#### `ON_FAILED_MESSAGE_CONSUMER`

Executed when message consumption fails due to an exception during processing.

Use this hook when you want to:

* log failures,
* send alerts,
* store error details,
* trigger retry or dead-letter related logic.

This hook is only executed for failed processing paths.


# Channel

## 📡 Channels

In the `@nestjstools/messaging` library, **Channels** represent the underlying **transport mechanisms** that deliver messages between microservices. Each channel is responsible for handling message transmission, reception, serialization, and routing to the appropriate handlers.

Channels abstract the communication details, allowing your application to work seamlessly whether using in-memory queues, RabbitMQ, or any other supported transport.

***

### 🔍 What Is a Channel?

A **Channel** is a pluggable component that:

* Sends and receives messages over a specific transport protocol
* Applies serialization and deserialization (normalizers)
* Manages queues, exchanges, topics, or other infrastructure details
* Supports middleware to process messages in the pipeline

***

### 🛠 Defining a Channel

Channels are configured within the MessagingModule via channel config classes. Each channel requires:

* A **unique name**
* Channel-specific options (connection info, queues, bindings)
* Optional **middlewares** and **normalizers**

#### Example: InMemory Channel

```ts
import { InMemoryChannelConfig } from '@nestjstools/messaging';

new InMemoryChannelConfig({
  name: 'my-channel',
  middlewares: [LoggingMiddleware],
});
```

#### Example: AMQP Channel

```ts
import { MessagingRabbitmqExtensionModule, RmqChannelConfig, ExchangeType } from '@nestjstools/messaging-rabbitmq-extension';

new RmqChannelConfig({
  name: 'amqp-command',
  connectionUri: 'amqp://guest:guest@localhost:5672/',
  exchangeName: 'my_app_command.exchange',
  bindingKeys: ['my_app.command.#'],
  exchangeType: ExchangeType.TOPIC,
  queue: 'my_app.command',
  autoCreate: true,
  enableConsumer: true,
});
```

***

### 🧩 Channel Configuration Options

| Option                     | Description                                                    |
| -------------------------- | -------------------------------------------------------------- |
| `name`                     | Unique identifier of the channel                               |
| `middlewares`              | List of middleware classes to apply on messages                |
| `normalizer`               | Serializer/deserializer to encode/decode messages              |
| Transport-specific options | Connection strings, queues, exchange names, binding keys, etc. |
| enableConsumer             | Enables or disables the RabbitMQ consumer for this channel     |

***

### 🔄 Channels and Buses

A **Bus** is connected to one or more channels. When a message is dispatched to a bus, it is forwarded to all its associated channels.

This design allows you to:

* Send messages to multiple transports simultaneously
* Mix different messaging protocols within the same application
* Configure channels independently for flexibility and scaling

***

### 🧩 Middleware and Normalizers on Channels

Each channel can have its own middleware stack and normalizers, enabling per-transport processing and custom encoding/decoding.

***

### 📦 Hook: `onChannelDestroy`

The `onChannelDestroy` The method is a lifecycle hook that is called when a channel instance is being destroyed. This is the appropriate place to clean up resources, close connections, or perform any teardown logic associated with your custom channel.

#### Purpose

Use `onChannelDestroy` to:

* Gracefully close external connections (e.g., sockets, database clients, message queues).
* Dispose of timers, intervals, or event listeners.
* Free up resources tied to the channel’s lifecycle.

```typescript
export class ExampleChannel extends Channel<ExampleChannelConfig> {
  constructor(config: ExampleChannelConfig) {
    super(config);
  }

  async onChannelDestroy(): Promise<void> {
    // Clean up logic here
    // For example, close an SQS client connection
    if (this.client) {
      await this.client.close();
    }
  }
}
```

#### Notes

* The method should return a `Promise<void>`.
* This hook is automatically called by the framework when the channel is being shut down.
* Always handle errors gracefully to avoid issues during shutdown.

***

### Benefits of Using Channels

| Feature                   | Benefit                                                       |
| ------------------------- | ------------------------------------------------------------- |
| 🔌 Transport Abstraction  | Switch between RabbitMQ, in-memory, or other protocols easily |
| ⚙️ Flexible Configuration | Fine-tune transport settings independently per channel        |
| 🧩 Extensible Middleware  | Add custom logic like logging, auth, or metrics per channel   |
| 🔄 Multi-Channel Support  | Dispatch messages across multiple transports simultaneously   |


# Broker integration


# RabbitMQ

## RabbitMQ Channel Integration

The `@nestjstools/messaging-rabbitmq-extension` provides seamless integration with **RabbitMQ** for asynchronous and synchronous message processing in NestJS applications.

### 📦 Installation

Install both core messaging and the RabbitMQ extension:

```bash
npm install @nestjstools/messaging @nestjstools/messaging-rabbitmq-extension
# or
yarn add @nestjstools/messaging @nestjstools/messaging-rabbitmq-extension
```

***

### 🧩 Basic Configuration Example

```ts
import { MessagingModule } from '@nestjstools/messaging';
import { MessagingRabbitmqExtensionModule, RmqChannelConfig, ExchangeType } from '@nestjstools/messaging-rabbitmq-extension';
import { SendMessageHandler } from './handlers/send-message.handler';

@Module({
  imports: [
    MessagingRabbitmqExtensionModule,
    MessagingModule.forRoot({
      messageHandlers: [SendMessageHandler],
      buses: [
        { name: 'message.bus', channels: ['my-channel'] },
        { name: 'command-bus', channels: ['amqp-command'] },
        { name: 'event-bus', channels: ['amqp-event'] },
      ],
      channels: [
        new InMemoryChannelConfig({ name: 'my-channel' }),
        new RmqChannelConfig({
          name: 'amqp-command',
          connectionUri: 'amqp://guest:guest@localhost:5672/',
          exchangeName: 'my_app_command.exchange',
          exchangeType: ExchangeType.TOPIC,
          queue: 'my_app.command',
          bindingKeys: ['my_app.command.#'],
          autoCreate: true,
        }),
        new RmqChannelConfig({
          name: 'amqp-event',
          connectionUri: 'amqp://guest:guest@localhost:5672/',
          exchangeName: 'my_app_event.exchange',
          exchangeType: ExchangeType.TOPIC,
          queue: 'my_app.event',
          bindingKeys: ['my_app_event.#'],
          autoCreate: true,
          avoidErrorsForNotExistedHandlers: true,
        }),
      ],
      debug: true,
    }),
  ],
})
export class AppModule {}
```

***

### 🛠 Exchange Types

| Exchange Type | Description                                                                         |
| ------------- | ----------------------------------------------------------------------------------- |
| `TOPIC`       | Route messages using wildcard-based routing keys (`my_app.command.#`).              |
| `DIRECT`      | Exact routing match. You must define matching `bindingKeys`.                        |
| `FANOUT`      | Broadcasts messages to all queues bound to the exchange, regardless of routing key. |

***

### 🔁 Cross-Language Messaging

You can publish messages from other services (non-NestJS apps) by following these rules:

1. **Send a Message** to the appropriate queue.
2. **Set Header:** `messaging-routing-key` should match the handler:

```ts
@MessageHandler('my_app_command.create_user')
```

***

### 🪦 Dead Letter Queue (DLQ) – How It Works

When `deadLetterQueueFeature: true` is enabled on an `AmqpChannelConfig`, the system automatically handles **failed messages** by routing them to a **dedicated "dead letter" queue** instead of discarding them or causing application crashes.

#### Behavior:

1. **Message Handling Fails**\
   If a message handler **throws an unhandled exception**, the message is not acknowledged (`nack`) and is redirected to the **DLQ**.
2. **DLQ Naming Convention**\
   The DLQ is created automatically and typically named by appending `dead_letter_queue`to the original queue name.\
   Example:\
   If your queue is `my_app.command`, the dead letter queue will be `my_app.command.dead_letter_queue`.
3. **Message Retention**\
   Failed messages remain in the DLQ until manually processed, examined, or retried.
4. **Retry Strategy**\
   You can **manually re-publish** messages from the DLQ back to the original exchange with the same routing key (or via tooling or scripts) when you're ready to retry.

***

#### 🔁 Example Use Case:

```ts
new RmqChannelConfig({
  name: 'amqp-command',
  connectionUri: 'amqp://guest:guest@localhost:5672/',
  exchangeName: 'my_app_command.exchange',
  exchangeType: ExchangeType.TOPIC,
  queue: 'my_app.command',
  bindingKeys: ['my_app.command.#'],
  autoCreate: true,
  deadLetterQueueFeature: true, // ✅ Enable DLQ
});
```

### 🔧 Configuration Table: `AmqpChannelConfig`

| Property                           | Description                                                                                                                                                                                                                                         | Default      |
| ---------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------ |
| `name`                             | Name of the channel (e.g., `'amqp-command'`).                                                                                                                                                                                                       | *(required)* |
| `connectionUri`                    | RabbitMQ connection URI (e.g., `'amqp://guest:guest@localhost:5672/'`).                                                                                                                                                                             | *(required)* |
| `exchangeName`                     | Exchange name in RabbitMQ.                                                                                                                                                                                                                          | *(required)* |
| `bindingKeys`                      | Routing keys for queue bindings (e.g., `['my_app.command.#']`).                                                                                                                                                                                     | `[]`         |
| `exchangeType`                     | Type of RabbitMQ exchange (`TOPIC`, `FANOUT`, `DIRECT`).                                                                                                                                                                                            | *(required)* |
| `queue`                            | Name of the queue to consume from.                                                                                                                                                                                                                  | *(required)* |
| `autoCreate`                       | Automatically create exchanges, queues, and bindings if missing.                                                                                                                                                                                    | `true`       |
| `enableConsumer`                   | Enable message consumption from this channel.                                                                                                                                                                                                       | `true`       |
| `enableWorker`                     | Enables the internal worker that processes messages. If `false`, messages are ignored.                                                                                                                                                              | `true`       |
| `avoidErrorsForNotExistedHandlers` | Skip errors when no handler exists for a routed message. Useful for optional event handlers.                                                                                                                                                        | `false`      |
| `middlewares`                      | Middleware pipeline for pre-processing messages.                                                                                                                                                                                                    | `[]`         |
| `normalizer`                       | Attach a normalizer for custom serialization (e.g., Protobuf, Base64).                                                                                                                                                                              | `undefined`  |
| `deadLetterQueueFeature`           | Enables capturing failed messages into a DLQ.                                                                                                                                                                                                       | `false`      |
| **`retryMessage`**                 | Number of times to retry a message before sending it to the dead letter queue. Only applicable if `deadLetterQueueFeature` is enabled.                                                                                                              |              |
| **`retryMessageTtl`**              | Time to live for retry messages in milliseconds. After this time, messages will be moved from the retry queue back to the main exchange for reprocessing. Only applicable if `retryMessage` is set.                                                 | `1000`       |
| **`forceRecreateRetryQueue`**      | Whether to forcefully recreate the retry queue on application startup. This can be useful during development to ensure a clean state, but should be used with caution in production environments as it will delete all messages in the retry queue. |              |

***

### ✉️ Custom Routing with AmqpMessageOptions

You can customize routing at dispatch time:

```ts
this.messageBus.dispatch(
  new RoutingMessage(
    new SendMessage('Hello Rabbit!'),
    'app.command.execute',
    new AmqpMessageOptions('exchange_name', 'rabbitmq_routing_key_to_queue')
  ),
);
```

### Mapping Messages in RabbitMQ Channels

RabbitMQ uses different **exchange types** to route messages based on routing keys and bindings. Here’s how message routing works for each exchange type in the context of messaging channels:

#### Topic Exchange

Topic exchanges route messages based on pattern-matching in the routing key.

* Use **wildcards** like `#` and `*` in your **binding keys** for flexible routing.
* **Example**: If you bind your queue with `my_app.command.#`, messages with routing keys such as `my_app.command.user.create` or `my_app.command.system.shutdown` will be routed to that queue.
* ✅ This is ideal for structured, hierarchical routing across many message types.

#### Direct Exchange

Direct exchanges use **exact matching** between the routing key and the binding key.

* Ensure that your queue has binding keys explicitly defined.
* If no binding key is provided, RabbitMQ defaults to the **routing key specified in the message handler**.
* Use this when you need precise, one-to-one message routing.

#### Fanout Exchange

Fanout exchanges **broadcast** messages to **all queues bound to the exchange**, **ignoring routing keys** entirely.

* Every bound queue receives the message.
* Best used for scenarios like logging, notifications, or pub-sub events where all consumers should receive the message.

### Retry messages flow

```
          publish (routingKey = "orders.created")
Producer  --------------------------------------------+
                                                       |
                                                       v
                                             +-------------------+
                                             | your.exchange    |
                                             | (topic/direct)    |
                                             +-------------------+
                                                       |
                                                       | bind: "orders.created" (or "#")
                                                       v
                                             +-------------------+
                                             | your_delay_queue       |
                                             | x-message-ttl=3000|
                                             | x-dead-letter-    |
                                             |   exchange=main.ex|
                                             +-------------------+
                                                       |
                             (after 3s TTL expires)     |
                                                       v
                                             +-------------------+
                                             | your.exchange     |
                                             | (topic)           |
                                             +-------------------+
                                                       |
                                                       | bind patterns:
                                                       |  "orders.*" / "orders.#"
                                                       v
                                             +-------------------+
                                             | your.queue        |
                                             +-------------------+
                                                       |
                                                       v
                                                   Consumer

```


# Redis

### 🧩 Redis Channel Integration

The Redis extension for `@nestjstools/messaging` enables your NestJS application to send and receive asynchronous messages through Redis-backed queues. This is especially useful for lightweight messaging setups or when using Redis as a shared infrastructure component.

***

#### 📦 Installation

Install the core and Redis messaging packages:

```bash
npm install @nestjstools/messaging @nestjstools/messaging-redis-extension
```

or

```bash
yarn add @nestjstools/messaging @nestjstools/messaging-redis-extension
```

***

#### ⚙️ Example Configuration

```ts
import { MessagingModule } from '@nestjstools/messaging';
import { MessagingRedisExtensionModule, RedisChannelConfig } from '@nestjstools/messaging-redis-extension';
import { InMemoryChannelConfig } from '@nestjstools/messaging/channels';
import { SendMessageHandler } from './handlers/send-message.handler';

@Module({
  imports: [
    MessagingRedisExtensionModule,
    MessagingModule.forRoot({
      messageHandlers: [SendMessageHandler],
      buses: [
        {
          name: 'message.bus',
          channels: ['my-channel'],
        },
        {
          name: 'command.bus',
          channels: ['redis-command'],
        },
        {
          name: 'event.bus',
          channels: ['redis-event'],
        },
      ],
      channels: [
        new InMemoryChannelConfig({
          name: 'my-channel',
          middlewares: [],
          avoidErrorsForNotExistedHandlers: true,
        }),
        new RedisChannelConfig({
          name: 'redis-command',
          queue: 'command-queue',
          connection: {
            host: '127.0.0.1',
            port: 6379,
          },
          middlewares: [],
          avoidErrorsForNotExistedHandlers: false,
        }),
        new RedisChannelConfig({
          name: 'redis-event',
          queue: 'event-queue',
          connection: {
            host: '127.0.0.1',
            port: 6379,
          },
          middlewares: [],
          avoidErrorsForNotExistedHandlers: true,
        }),
      ],
      debug: true,
    }),
  ],
})
export class AppModule {}
```


# Google PubSub

### Google Pub/Sub Channel Integration

The Google Pub/Sub extension for `@nestjstools/messaging` enables cloud-native, event-driven communication across distributed systems. It’s ideal for scalable, serverless-friendly applications that rely on Google Cloud infrastructure.

***

#### 📦 Installation

Install the core messaging package and the Google Pub/Sub extension:

```bash
npm install @nestjstools/messaging @nestjstools/messaging-google-pubsub-extension
```

or

```bash
yarn add @nestjstools/messaging @nestjstools/messaging-google-pubsub-extension
```

***

#### ⚙️ Example Configuration

```ts
import { MessagingModule } from '@nestjstools/messaging';
import { PubSubChannelConfig, MessagingGooglePubSubExtensionModule } from '@nestjstools/messaging-google-pubsub-extension';
import { InMemoryChannelConfig } from '@nestjstools/messaging/channels';
import { SendMessageHandler } from './handlers/send-message.handler';

@Module({
  imports: [
    MessagingGooglePubSubExtensionModule,
    MessagingModule.forRoot({
      messageHandlers: [SendMessageHandler],
      buses: [
        {
          name: 'default.bus',
          channels: ['in-memory', 'pubsub-events'],
        },
      ],
      channels: [
        new InMemoryChannelConfig({
          name: 'in-memory',
          avoidErrorsForNotExistedHandlers: true,
        }),
        new PubSubChannelConfig({
          name: 'pubsub-events',
          projectId: 'your-gcp-project-id',
          topic: 'your-topic-name',
          subscription: 'your-subscription-name',
          middlewares: [],
          autoCreate: true,
          enableSubscriber: true,
          avoidErrorsForNotExistedHandlers: true,
        }),
      ],
      debug: true,
    }),
  ],
})
export class AppModule {}
```

***

#### 🛠️ PubSubChannelConfig Properties

| Property                           | Description                                                               | Default Value |
| ---------------------------------- | ------------------------------------------------------------------------- | ------------- |
| `name`                             | The channel name (e.g., `'pubsub-events'`).                               | —             |
| `projectId`                        | Google Cloud project ID.                                                  | —             |
| `topic`                            | The Pub/Sub topic to publish messages to.                                 | —             |
| `subscription`                     | Subscription name to consume messages from.                               | —             |
| `middlewares`                      | Array of middlewares applied to incoming/outgoing messages.               | `[]`          |
| `autoCreate`                       | Automatically create topic/subscription if they do not exist.             | `true`        |
| `enableSubscriber`                 | Enables the background subscriber to listen to incoming Pub/Sub messages. | `true`        |
| `avoidErrorsForNotExistedHandlers` | Silently skips messages without a matching handler.                       | `false`       |

***

#### 🌐 Key Features

* **Cloud-Native Messaging**: Leverages Google Pub/Sub for reliable, asynchronous communication across microservices.
* **Auto Resource Provisioning**: Automatically creates missing topics and subscriptions when `autoCreate` is enabled.
* **Efficient Scaling**: Suitable for horizontally scaled environments (e.g., Kubernetes, Cloud Run).
* **Graceful Error Handling**: Skip errors when no handler exists by enabling `avoidErrorsForNotExistedHandlers`.
* **Subscriber Control**: Toggle subscriber activation with `enableSubscriber`.


# Amazon SQS

### 📨 Amazon SQS Channel Integration

This extension enables seamless integration between `@nestjstools/messaging` and **Amazon Simple Queue Service (SQS)**. It supports both cloud-based and local queue setups (via ElasticMQ) for reliable, scalable, and distributed messaging.

***

#### 📦 Installation

```bash
npm install @nestjstools/messaging @nestjstools/messaging-amazon-sqs-extension
```

or

```bash
yarn add @nestjstools/messaging @nestjstools/messaging-amazon-sqs-extension
```

***

#### ⚙️ Example Configuration

```ts
import { Module } from '@nestjs/common';
import { MessagingModule } from '@nestjstools/messaging';
import { MessagingAmazonSQSExtensionModule, AmazonSqsChannelConfig } from '@nestjstools/messaging-amazon-sqs-extension';
import { SendMessageHandler } from './handlers/send-message.handler';

@Module({
  imports: [
    MessagingAmazonSQSExtensionModule,
    MessagingModule.forRoot({
      messageHandlers: [SendMessageHandler],
      buses: [
        {
          name: 'sqs-event.bus',
          channels: ['sqs-event'],
        },
      ],
      channels: [
        new AmazonSqsChannelConfig({
          name: 'sqs-event',
          region: 'us-east-1',
          queueUrl: 'http://localhost:9324/queue/test_queue', // ElasticMQ for local use
          autoCreate: true,
          enableConsumer: true,
          credentials: {
            accessKeyId: 'x',
            secretAccessKey: 'x',
          },
          maxNumberOfMessages: 3,
          visibilityTimeout: 10,
          waitTimeSeconds: 5,
        }),
      ],
      debug: true,
    }),
  ],
})
export class AppModule {}
```

***

#### 🛠️ AmazonSqsChannelConfig Properties

| Property              | Description                                                        | Default Value |
| --------------------- | ------------------------------------------------------------------ | ------------- |
| `name`                | Name of the SQS channel (e.g., `'sqs-event'`).                     | —             |
| `region`              | AWS region for the queue (e.g., `'us-east-1'`).                    | —             |
| `queueUrl`            | Full URL of the SQS queue.                                         | —             |
| `credentials`         | Optional AWS credentials (accessKeyId & secretAccessKey).          | —             |
| `enableConsumer`      | Whether to enable consuming messages from this queue.              | `true`        |
| `autoCreate`          | Automatically create the queue if it doesn't exist.                | `true`        |
| `maxNumberOfMessages` | Number of messages to fetch in a single poll.                      | `1`           |
| `visibilityTimeout`   | Time (seconds) to hide a message after retrieval.                  | `20`          |
| `waitTimeSeconds`     | Duration (seconds) the consumer waits for messages (long polling). | `0`           |

***

#### 🌐 Cross-Language Communication

To integrate with external (non-NestJS) systems:

* **Publish a message** to the SQS queue.
* **Set the `messagingRoutingKey` header** to match your NestJS handler:

```ts
@MessageHandler('my_app_command.create_user') // routing key to use in SQS message attributes
```


# Nats

## Messaging with NATS

### ⚙️ Installation

```bash
npm install @nestjstools/messaging @nestjstools/messaging-nats-extension 
```

or

```bash
yarn add @nestjstools/messaging @nestjstools/messaging-nats-extension
```

### 📦 Overview

This guide demonstrates how to integrate NATS (and NATS JetStream) into a NestJS application using `@nestjstools/messaging` and the `messaging-nats-extension`.

We cover:

* Basic NATS setup
* Using JetStream
* Message dispatch and handling
* Cross-language messaging
* Routing strategies
* Configuration options

***

### ⚙️ Basic NATS Configuration

```ts
import { Module } from '@nestjs/common';
import { MessagingModule } from '@nestjstools/messaging';
import { MessagingNatsExtensionModule, NatsChannelConfig } from '@nestjstools/messaging-nats-extension';

@Module({
  imports: [
    MessagingNatsExtensionModule,
    MessagingModule.forRoot({
      buses: [
        {
          name: 'nats-message.bus',
          channels: ['nats-message'],
        },
      ],
      channels: [
        new NatsChannelConfig({
          name: 'nats-message',
          enableConsumer: true,
          connectionUris: ['nats://localhost:4222'],
          subscriberName: 'nats-core',
        }),
      ],
      debug: true,
    }),
  ],
})
export class AppModule {}
```

***

### 🚀 JetStream Configuration

```ts
import { Module } from '@nestjs/common';
import { MessagingModule } from '@nestjstools/messaging';
import { MessagingNatsExtensionModule, NatsJetStreamChannelConfig } from '@nestjstools/messaging-nats-extension';

@Module({
  imports: [
    MessagingNatsExtensionModule,
    MessagingModule.forRoot({
      buses: [
        {
          name: 'nats-message.bus',
          channels: ['nats-channel-jetstream'],
        },
      ],
      channels: [
        new NatsJetStreamChannelConfig({
          name: 'nats-channel-jetstream',
          connectionUris: ['nats://localhost:4222'],
          enableConsumer: true,
          streamConfig: {
            streamName: 'event-steam',
            deliverSubjects: ['my_app_command.*'],
            autoUpdate: true,
          },
          consumerConfig: {
            durableName: 'nats-durable_name',
            subject: 'my_app_command.*',
            autoUpdate: true,
          },
        }),
      ],
      debug: true,
    }),
  ],
})
export class AppModule {}
```

***

### 📤 Dispatching Messages

Use a controller to send messages through the bus.

```ts
import { Controller, Get } from '@nestjs/common';
import { CreateUser } from './application/command/create-user';
import { IMessageBus, MessageBus, RoutingMessage } from '@nestjstools/messaging';

@Controller()
export class AppController {
  constructor(
    @MessageBus('nats-message.bus') private natsMessageBus: IMessageBus,
  ) {}

  @Get('/nats')
  createUser(): string {
    this.natsMessageBus.dispatch(
      new RoutingMessage(new CreateUser('John FROM Nats'), 'my_app_command.create_user'),
    );
    return 'Message sent';
  }
}
```

***

### 📥 Handling Messages

Create a handler that listens to a specific routing key:

```ts
import { CreateUser } from '../create-user';
import {
  IMessageHandler,
  MessageHandler,
} from '@nestjstools/messaging';

@MessageHandler('my_app_command.create_user')
export class CreateUserHandler implements IMessageHandler<CreateUser> {
  async handle(message: CreateUser): Promise<void> {
    console.log(message);
    // Your logic here
  }
}
```

***

### 🌐 Cross-Language Communication

To interact with NestJS handlers from external services (e.g., written in Go, Python, etc.):

1. **Publish a message to the queue**
2. **Include the `messaging-routing-key` header**

```ts
@MessageHandler('my_app_command.create_user') // <-- Use this as the routing key
```

3. That’s it! The NestJS app will route the message to the correct handler.

***

### 🧭 Routing Strategy

Routing is determined by the `subscriberName` or `deliverSubjects`.

#### Static Routing

If `subscriberName` is a concrete subject:

```ts
subscriberName = 'order.created';
// Message will be sent to 'order.created'
```

#### Wildcard Routing

If `subscriberName` uses a wildcard:

```ts
subscriberName = 'order.*';
message.messageRoutingKey = 'order.created';
// Message will be sent to 'order.created'
```

**JetStream Example**

```ts
subject = 'order.*'; // from consumer
message.messageRoutingKey = 'order.created';
// The message will be published to 'order.created'
```

***

### ⚙️ Configuration Options

#### `NatsChannelConfig`

| Property         | Description                                          |
| ---------------- | ---------------------------------------------------- |
| `name`           | Name of the NATS channel (e.g., `'nats-message'`)    |
| `enableConsumer` | Enable message consumption                           |
| `connectionUris` | NATS server URIs (e.g., `['nats://localhost:4222']`) |
| `subscriberName` | Unique identifier for the subscriber                 |

> 📝 **Note:** JetStream and NATS offer extensive configurations. If this setup doesn't suit your needs, you can fork and customize the base channel classes provided in this package.


# Azure Service Bus

Absolutely! Here's a great starting structure for a **GitBook page** explaining your Azure Service Bus integration.

***

## &#x20;Azure Service Bus Integration

> This guide explains how to configure and use Azure Service Bus in your messaging-based application using `@nestjstools/messaging`.

***

### 📦 Installation

Make sure you’ve installed the following dependencies:

```bash
npm install @nestjstools/messaging @nestjstools/messaging-azure-service-bus-extension
or
yarn add @nestjstools/messaging @nestjstools/messaging-azure-service-bus-extension
```

***

### 🛠️ Channel Configuration

#### Basic Setup queue

```ts
channels: [
  new AzureServiceBusChannelConfig({
    name: 'azure-channel',
    connectionString: 'Endpoint=sb://your-namespace.servicebus.windows.net/;SharedAccessKeyName=...;',
    queue: 'your-queue-name',
    autoCreate: false, // Requires admin permission
    enableConsumer: true,
  }),
],
```

#### Topic/Subscription Example

```ts
channels: [
  new AzureServiceBusChannelConfig({
    name: 'azure-pubsub-channel',
    connectionString: 'Endpoint=sb://your-namespace.servicebus.windows.net/;SharedAccessKeyName=...;',
    topic: 'your-topic',
    subscription: 'your-subscription',
    mode: Mode.TOPIC,
    autoCreate: true,
    enableConsumer: true,
  }),
],
```

***

### ⚙️ Config Options

| Property           | Description                                                       | Default   |
| ------------------ | ----------------------------------------------------------------- | --------- |
| `name`             | Internal channel name                                             | —         |
| `connectionString` | Azure Service Bus connection string                               | —         |
| `queue`            | Queue name (required for `Mode.QUEUE`)                            | —         |
| `topic`            | Topic name (required for `Mode.TOPIC`)                            | —         |
| `subscription`     | Subscription name (required for `Mode.TOPIC`)                     | —         |
| `mode`             | `'queue'` or `'topic'`                                            | `'queue'` |
| `autoCreate`       | Auto-create queue/topic/subscription (requires admin permissions) | `false`   |
| `enableConsumer`   | Enable message receiving                                          | `true`    |

***

### 📤 Dispatching a Message

```ts
import { Controller, Get } from '@nestjs/common';
import { CreateUser } from './application/command/create-user';
import { IMessageBus, MessageBus, RoutingMessage } from '@nestjstools/messaging';

@Controller()
export class AppController {
  constructor(
    @MessageBus('azure.bus') private azureMessageBus: IMessageBus,
  ) {}

  @Get('/azure')
  createUser(): string {
    this.azureMessageBus.dispatch(
      new RoutingMessage(
        new CreateUser('John FROM Azure bus'),
        'my_app_command.create_user',
      ),
    );

    return 'Message sent';
  }
}
```

***

### ✅ Tips

* `autoCreate` works **only if** the channel has `enableConsumer = true` and the connection string has **management permissions**.
* Use `mode: Mode.TOPIC` only when using `topic` and `subscription`.


# Best practice


# CQRS based on RabbitMQ

## CQRS in NestJS Using RabbitMQ

**CQRS (Command Query Responsibility Segregation)** is a design pattern that decouples the read (query) and write (command) sides of your application. By combining this pattern with **RabbitMQ** and NestJS’s powerful modular system, we can build a robust and scalable architecture for asynchronous processing and eventual consistency.

This article explores how to implement a CQRS system in NestJS using the `@nestjstools/messaging` library and a RabbitMQ-backed channel.

***

### 🧱 Why CQRS with Messaging?

* **Separation of concerns**: Commands (writes) and queries (reads) evolve independently.
* **Scalability**: Commands can be processed asynchronously in separate services.
* **Event-driven architecture**: Events can propagate across distributed systems.
* **Reliability**: RabbitMQ queues offer retry logic, durability, and fault tolerance.

***

### 🛠 Messaging Setup with RabbitMQ

We define a `MessagingWrapperModule` that registers message buses and connects them to appropriate channels:

```ts
@Module({})
export class MessagingWrapperModule {
  static forRoot(enableConsumer: boolean): DynamicModule {
    return {
      imports: [
        MessagingRabbitmqExtensionModule,
        MessagingModule.forRoot({
          buses: [
            { name: 'sync-command.bus', channels: ['sync-channel'] },
            { name: 'command.bus', channels: ['amqp-command.channel'] },
            { name: 'event.bus', channels: ['amqp-event.channel'] },
          ],
          channels: [
            new InMemoryChannelConfig({ name: 'sync-channel' }),
            new AmqpChannelConfig({
              name: 'amqp-command.channel',
              connectionUri: 'amqp://guest:guest@localhost:5672/',
              exchangeName: 'book_shop.exchange',
              bindingKeys: ['book_shop.#'],
              exchangeType: ExchangeType.TOPIC,
              queue: 'book_shop.command',
              autoCreate: true,
              enableConsumer,
            }),
            new AmqpChannelConfig({
              name: 'amqp-event.channel',
              connectionUri: 'amqp://guest:guest@localhost:5672/',
              exchangeName: 'book_shop.exchange',
              bindingKeys: ['book_shop.#'],
              exchangeType: ExchangeType.TOPIC,
              queue: 'book_shop.event',
              autoCreate: true,
              enableConsumer,
            }),
          ],
          debug: true,
        }),
      ],
      providers: [
        {
          provide: Service.CommandBus,
          useFactory: (bus: IMessageBus) => new InternalMessageBus(bus),
          inject: ['command.bus'],
        },
        {
          provide: Service.EventBus,
          useFactory: (bus: IMessageBus) => new InternalMessageBus(bus),
          inject: ['event.bus'],
        },
        {
          provide: Service.SyncCommandBus,
          useFactory: (bus: IMessageBus) => new InternalMessageBus(bus),
          inject: ['sync-command.bus'],
        },
      ],
      exports: [
        Service.CommandBus,
        Service.EventBus,
        Service.SyncCommandBus,
      ],
      module: MessagingWrapperModule,
      global: true,
    };
  }
}
```

***

### 📦 InternalMessageBus Wrapper

```ts
export class InternalMessageBus {
  constructor(private readonly messageBus: IMessageBus) {}

  async dispatch(message: object, routingKey: string): Promise<void> {
    this.messageBus.dispatch(new RoutingMessage(message, routingKey));
    return Promise.resolve();
  }
}
```

This class acts as a lightweight abstraction over the raw `IMessageBus`, giving you a consistent and simplified interface to dispatch messages.

***

### 🧩 Dependency Injection

Use the following helpers to inject your buses where needed:

```ts
import { Inject } from '@nestjs/common';
import { Service } from '@messaging-wrapper/messaging-wrapper/dependency-injection/service';

export const SyncCommandBus = Inject(Service.SyncCommandBus);
export const CommandBus = Inject(Service.CommandBus);
export const EventBus = Inject(Service.EventBus);
```

The service identifiers are managed through an enum:

```ts
export enum Service {
  SyncCommandBus = 'MCommandBus',
  CommandBus = 'MSyncCommandBus',
  EventBus = 'MSyncEventBus',
}
```

***

### ✅ Example Usage

```ts
@Controller('/orders')
export class OrderController {
  constructor(@CommandBus private readonly commandBus: InternalMessageBus) {}

  @Get('/simulate')
  simulate(): string {
    this.commandBus.dispatch(
      new CompleteOrder('order-uuid-123', 'Star Wars: The New Galactic', 2),
      'book_shop.command.complete_order',
    );
    return 'ok';
  }
}
```

***

### 🧪 Final Thoughts

With RabbitMQ and CQRS in NestJS:

* Commands and events are routed across channels and buses.
* Consumers process messages reliably in background workers.
* The architecture supports distributed, event-driven patterns with ease.

This setup makes your system **more scalable**, **maintainable**, and **resilient**—ideal for microservices, modular monoliths, and modern DDD architectures.


# Create wrapper class for Message Bus

### 🚀 Messaging Wrapper Module

This wrapper provides a reusable, centralized setup for working with the `@nestjstools/messaging` library. It encapsulates **sync-command** bus into injectable services that can be easily used across your application.

***

#### 📦 Purpose

Instead of configuring your message buses and channels in every module, this wrapper:

* Exposes  `SyncCommandBus` as injectable services
* Ensures clean separation of sync/async responsibilities
* Keeps your controller/service code lean and focused

***

```typescript
import { IMessageBus, RoutingMessage } from '@nestjstools/messaging';

/**
 * A simplified interface for dispatching messages through a configured message bus.
 * Encapsulates routing logic to keep controllers and services clean.
 */
export class InternalMessageBus {
  constructor(private readonly messageBus: IMessageBus) {}

  /**
   * Dispatch a message using the given routing key.
   *
   * @param message The payload (command/event/etc.)
   * @param routingKey The message routing key used by the bus
   */
  async dispatch(message: object, routingKey: string): Promise<void> {
    await this.messageBus.dispatch(new RoutingMessage(message, routingKey));
  }
}
```

#### 🧱 Module Setup

```ts
@Module({})
export class MessagingWrapperModule {
  static forRoot(): DynamicModule {
    return {
      imports: [
        MessagingModule.forRoot({
          buses: [
            { name: 'sync-command.bus', channels: ['sync-channel'] },
          ],
          channels: [
            new InMemoryChannelConfig({
              name: 'sync-channel',
            }),
      
          ],
          debug: true,
        }),
      ],
      providers: [
        {
          provide: 'MSyncCommandBus',
          useFactory: (bus: IMessageBus) => new MessageBus(bus),
          inject: ['sync-command.bus'],
        },
      ],
      exports: [
        'MSyncCommandBus',
      ],
      module: MessagingWrapperModule,
      global: true,
    };
  }
}
```

***

#### 🧩 Dependency Injection Tokens

These are helper tokens to inject the appropriate bus:

```ts
import { Inject } from '@nestjs/common';
import { Service } from '@messaging-wrapper/messaging-wrapper/dependency-injection/service';

export const SyncCommandBus = Inject('MSyncCommandBus');
```

***

#### 🧪 Usage in a Controller

Example of dispatching a command using the injected `@SyncCommandBus` :

```ts
@Controller('/orders')
export class OrderController {
  constructor(@SyncCommandBus private readonly commandBus: MessageBus) {}

  @Get('/simulate_complete')
  simulateCompleteOrder(): string {
    this.commandBus.dispatch(
      new CompleteOrder('uuid-order-123', 'Star Wars: The New Galactic', 2),
      'complete.order',
    );

    return 'ok';
  }
}
```


# Consumer as the background process

> ⚠️ **Before You Continue**
>
> It's recommended to read the [Bootstrap ](/best-practice/bootstrap-http-worker-mode)Guide first.\
> It provides a higher-level, smarter approach for managing messaging in both HTTP server and worker (microservice) modes.
>
> This page is useful if you're configuring consumers manually or outside the bootstrap utility.

### 🧾 Consumer as a Background Process

In your messaging configuration, you have the ability to run certain channels in **background consumer mode**, meaning the application will continuously poll and process messages from a queue (e.g., RabbitMQ).

This behavior is **explicitly controlled** using the `enableConsumer` flag inside the channel configuration.

***

#### ✅ Enabling or Disabling Consumers

Each channel can individually opt into message consumption:

```ts
new AmqpChannelConfig({
  name: 'amqp-command.channel',
  connectionUri: 'amqp://guest:guest@localhost:5672/',
  exchangeName: 'book_shop.exchange',
  bindingKeys: ['book_shop.#'],
  exchangeType: ExchangeType.TOPIC,
  queue: 'book_shop.command',
  enableConsumer: process.env.CONSUMER_ENABLED === 'true' ?? false, // ← starts the background consumer
})
```

Setting `enableConsumer: true` allows the messaging module to **automatically listen to that queue** and handle incoming messages to the proper handlers in your NestJS app.

***

### ⚙️ Running the Consumer as a Dedicated Microservice

In addition to toggling `enableConsumer` in your channel config, you can fully isolate message consumption into its own process using **NestJS microservice mode**.

This is useful when you want to **separate HTTP requests from background processing**, scale them independently, or run consumers in isolated environments (e.g., workers).

***

#### 🔁 From HTTP App to Microservice Consumer

Your standard `main.ts` for a typical HTTP NestJS app might look like this:

```ts
// Standard HTTP entry point
process.env.CONSUMER_ENABLED = 'false';
import { NestFactory } from '@nestjs/core';
import { BookModule } from './book.module';

async function bootstrap() {
  const app = await NestFactory.create(BookModule.forRoot());
  await app.listen(process.env.port ?? 3000);
}
bootstrap();
```

You can **create** `worker.ts` **a microservice-based consumer** by doing the following:

#### ✅ Microservice Consumer Entry

```ts
// microservice-main.ts
process.env.CONSUMER_ENABLED = 'true'; // ensures consumers are enabled via config

import { NestFactory } from '@nestjs/core';
import { Transport, MicroserviceOptions } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    AppModule,
    {
      transport: Transport.TCP,
    },
  );
  await app.listen();
}
bootstrap();
```


# Bootstrap (Http, Worker mode)

#### 📘 Introduction

> A simple wrapper to speed up messaging-based app setup in NestJS.

[`@nestjstools/messaging-bootstrap`](https://www.npmjs.com/package/@nestjstools/messaging-bootstrap) It is a lightweight utility that simplifies bootstrapping a messaging-enabled NestJS application.

Built on top of [`@nestjstools/messaging`](https://www.npmjs.com/package/@nestjstools/messaging), it offers two main modes:

* An **HTTP server** with integrated messaging publisher capabilities
* A **dedicated worker/microservice** that runs only messaging consumers

***

#### ⚙️ Installation

```bash
yarn add @nestjstools/messaging-bootstrap @nestjs/microservices
```

***

#### 🚀 Quick Start

Create a file `messaging-confg.ts`

```typescript
// messaging-confg.ts

import { MessagingRabbitmqExtensionModule } from '@nestjstools/messaging-rabbitmq-extension';
import { AmqpChannelConfig, ExchangeType } from '@nestjstools/messaging';
import { MessagingModuleConfig } from '@nestjstools/messaging-bootstrap';

export const Config: MessagingModuleConfig = {
  extensions: [
    MessagingRabbitmqExtensionModule
  ],
  buses: [
    { name: 'command.bus', channels: ['async-command'] }
  ],
  channels: [
    new AmqpChannelConfig({
      name: 'async-command',
      connectionUri: 'amqp://localhost',
      exchangeName: 'command.exchange',
      bindingKeys: ['command.#'],
      exchangeType: ExchangeType.TOPIC,
      queue: 'app.command',
      enableConsumer: false,
    }),
  ],
};

```

**✅ HTTP Server with Messaging Publisher**

```ts
// main.ts
import { AppModule } from './app.module';
import { ConsoleLogger } from '@nestjs/common';
import { MessagingBootstrap } from '@nestjstools/messaging-bootstrap';
import { Config } from './messaging-confg';

async function bootstrap() {
  const app = await MessagingBootstrap.createNestApplicationWithMessaging(AppModule, {
    messaging: Config,
    nestApplicationOptions: {
      logger: new ConsoleLogger({ json: true }),
    },
  });
  await app.listen(3000);
}
bootstrap();

```

**✅ Worker/Microservice with Consumers**

```ts
// worker.ts
import { AppModule } from './app.module';
import { ConsoleLogger } from '@nestjs/common';
import { MessagingBootstrap } from '@nestjstools/messaging-bootstrap';
import { Config } from './messaging-confg';

async function bootstrap() {
  const app = await MessagingBootstrap.createNestMicroserviceWithMessagingConsumer(AppModule, {
    messaging: Config,
    nestMicroserviceOptions: {
      logger: new ConsoleLogger({ json: true }),
    },
  });
  await app.listen();
}
bootstrap();
```

***

#### 🛠️ 4. Configuration

All options are passed into the bootstrap function:

**`messaging` options:**

* `extensions`: array of messaging extension modules (e.g. RabbitMQ) (optional)
* `buses`: define one or more message buses
* `channels`: define channel configs (e.g. AMQP)

**`nestApplicationOptions` /** `nestMicroserviceOptions`**:**

* Directly passed to `NestFactory.create` or `createMicroservice`

***

#### 📄 5. Examples

**Shared Configuration Pattern**

```ts
export const messagingConfig = {
  extensions: [MessagingRabbitmqExtensionModule],
  buses: [{ name: 'command.bus', channels: ['async-command'] }],
  channels: [
    new AmqpChannelConfig({
      name: 'async-command',
      connectionUri: 'amqp://localhost',
      exchangeName: 'command.exchange',
      queue: 'app.command',
      bindingKeys: ['command.#'],
      exchangeType: ExchangeType.TOPIC,
      deadLetterQueueFeature: true,
      autoCreate: true,
      enableConsumer: false, // worker will auto-enable it
    }),
  ],
};
```

Then reuse in both `main.ts` and `worker.ts`.

***

#### ⚠️ 7. Common Pitfalls

**❌ Do not use `MessagingModule.forRoot()` manually**

> The bootstrap library already handles it for you. Including it manually will lead to **duplicate initialization** and **unexpected bugs**.

Correct usage:

```ts
// DO NOT DO THIS!
@Module({
  imports: [
    MessagingModule.forRoot({...}) // ❌ DON'T DO THIS
  ]
})
export class AppModule {}
```

Instead, let the bootstrap function handle it automatically.


# Create custom channel

`@nestjstools/messaging` allows you to create **custom transports** by implementing your own **Channel, MessageBus, and Consumer**.

This makes it possible to integrate the messaging system with external technologies such as:

* RabbitMQ
* Redis
* NATS
* Google Pub/Sub
* AWS SQS
* **or any other messaging system**

A custom channel acts as the **transport layer** responsible for delivering and consuming messages.

***

## 1. Create a ChannelConfig

`ChannelConfig` stores configuration required to establish a connection to the messaging system.

```ts
import { ChannelConfig } from '@nestjstools/messaging';

export class YourChannelConfig extends ChannelConfig {
  public readonly connectionUri: string;
  public readonly queue: string;

  constructor({
    name,
    connectionUri,
    queue,
    avoidErrorsForNotExistedHandlers,
    middlewares,
    enableConsumer,
    normalizer,
  }: {
    name: string;
    connectionUri: string;
    queue: string;
    avoidErrorsForNotExistedHandlers?: boolean;
    middlewares?: object[];
    enableConsumer?: boolean;
    normalizer?: object;
  }) {
    super(
      name,
      avoidErrorsForNotExistedHandlers,
      middlewares,
      enableConsumer,
      normalizer,
    );

    this.connectionUri = connectionUri;
    this.queue = queue;
  }
}
```

Typical configuration may include:

* connection settings
* channel name
* middleware configuration
* transport-specific options

***

## 2. Create a Channel

The `Channel` acts as the **data source layer** and manages the connection to the external service.

```ts
import { Channel } from '@nestjstools/messaging';

export class YourChannel extends Channel<YourChannelConfig> {
  private client: unknown;

  constructor(public readonly config: YourChannelConfig) {
    super(config);

    // initialize transport client here for example RabbitMQ connection
    this.client = {};
  }

  getClient(): unknown {
    return this.client;
  }

  async onChannelDestroy(): Promise<void> {
    // close connection here
  }
}
```

This class can manage:

* connections
* transport resources

***

## 3. Create a ChannelFactory

The `ChannelFactory` creates channel instances and integrates them with NestJS dependency injection.

```ts
import { Injectable } from '@nestjs/common';
import {
  ChannelFactory,
  IChannelFactory,
  Channel,
} from '@nestjstools/messaging';

@Injectable()
@ChannelFactory(YourChannel)
export class YourChannelFactory implements IChannelFactory<YourChannelConfig> {
  create(channelConfig: YourChannelConfig): Channel {
    return new YourChannel(channelConfig);
  }
}
```

***

## 4. Create a MessageBus

The `MessageBus` is responsible for dispatching messages to the transport layer.

```ts
import {
  IMessageBus,
  RoutingMessage,
  MessageResponse,
} from '@nestjstools/messaging';

export class YourMessageBus implements IMessageBus {
  constructor(private readonly yourChannel: YourChannel) {}

  async dispatch(message: RoutingMessage): Promise<MessageResponse | void> {

    // Example RabbitMQ logic:
    //
    // Get AMQP channel wrapper from your channel implementation
    // const channelWrapper = this.yourChannel.createChannelWrapper();
    //
    // Prepare message payload
    // const payload = Buffer.from(JSON.stringify(message));
    //
    // Publish message to exchange
    // await channelWrapper.publish(
    //   this.yourChannel.config.exchangeName,   // exchange name
    //   message.routingKey ?? '',               // routing key
    //   payload,                                // message body
    //   {
    //     persistent: true,                     // survive broker restart
    //     contentType: 'application/json',
    //     headers: {
    //       messageType: message.constructor.name,
    //     },
    //   }
    // );
  }
}
```

This is where you integrate with your messaging system (RabbitMQ, Redis, etc.).

***

## 5. Create a MessageBusFactory

The `MessageBusFactory` creates instances of your message bus.

```ts
import { Injectable } from '@nestjs/common';
import {
  MessageBusFactory,
  IMessageBusFactory,
  IMessageBus,
} from '@nestjstools/messaging';

@Injectable()
@MessageBusFactory(YourChannel)
export class YourMessageBusFactory implements IMessageBusFactory<YourChannel> {
  create(channel: YourChannel): IMessageBus {
    return new YourMessageBus(channel);
  }
}
```

***

## 6. Create a Consumer

A consumer reads messages from the transport and dispatches them to handlers inside the application.

```ts
import { Injectable } from '@nestjs/common';
import {
  MessageConsumer,
  IMessagingConsumer,
  ConsumerMessageBus,
  ConsumerMessage,
  ConsumerDispatchedMessageError,
} from '@nestjstools/messaging';

@Injectable()
@MessageConsumer(YourChannel)
export class YourMessagingConsumer implements IMessagingConsumer<YourChannel> {

  async consume(
    dispatcher: ConsumerMessageBus,
    channel: YourChannel,
  ): Promise<void> {

    // 1. Connect to transport (RabbitMQ, Redis, etc.)
    // const connection = await connect(channel.config.connectionUri);

    // 2. Subscribe to queue / topic
    // connection.consume(channel.config.queue, async (rawMessage) => {

    // 3. Deserialize payload
    // const payload = JSON.parse(rawMessage.content.toString());

    // 4. Dispatch message into messaging system
    // await dispatcher.dispatch(
    //   new ConsumerMessage(payload, rawMessage.routingKey)
    // );

    // 5. Acknowledge message
    // rawMessage.ack();

    // });

  }

  async onError(
    errored: ConsumerDispatchedMessageError,
    channel: YourChannel,
  ): Promise<void> {

    // Handle message processing errors.
    // Typical strategies include:
    //
    // - retrying the message
    // - sending the message to a dead-letter queue
    // - logging the failure

  }
}

```

The consumer should:

* read messages from the messaging system
* dispatch them to application handlers
* handle processing errors

***

## 7. Custom MessageOptions (Optional)

You can define custom message options for your transport and build custom logic like adding headers etc.

```ts
import { MessageOptions, Middleware } from '@nestjstools/messaging';

export class YourMessageOptions implements MessageOptions {
  constructor(public readonly middlewares: Middleware[] = []) {}
}
```

***

## Registering Providers

Classes decorated with `@Injectable()` must be registered as providers in your NestJS module.

```ts
import { Module } from '@nestjs/common';

@Module({
  providers: [
    YourChannelFactory,
    YourMessageBusFactory,
    YourMessagingConsumer,
  ],
})
export class MessagingExtensionModule {}
```


# Configuration

`MessagingModule` can be configured using `MessagingModule.forRoot()`.

This configuration defines:

* **buses** – logical message buses used by your application
* **channels** – transports responsible for delivering messages
* **debug and logging behavior**
* **consumer execution rules**

Example configuration:

```ts
import { MessagingModule, InMemoryChannelConfig } from '@nestjstools/messaging';

MessagingModule.forRoot({
  buses: [
    {
      name: 'event.message-bus',
      channels: ['in-memory-channel'],
    },
  ],
  channels: [
    new InMemoryChannelConfig({
      name: 'in-memory-channel',
    }),
  ],
  debug: false,
  logging: true,
});
```

***

## MessagingModule.forRoot Options

| Property                   | Description                                                                                                               | Default      |
| -------------------------- | ------------------------------------------------------------------------------------------------------------------------- | ------------ |
| `buses`                    | Array of message buses defining routing and processing of messages.                                                       | `[]`         |
| `channels`                 | Array of channel configurations used by the message buses.                                                                | `[]`         |
| `debug`                    | Enables debug mode with additional logs useful during development.                                                        | `false`      |
| `logging`                  | Enables logging for bus activity (e.g., message dispatching).                                                             | `true`       |
| `customLogger`             | Custom logger instance implementing `MessagingLogger`.                                                                    | `NestLogger` |
| `forceDisableAllConsumers` | Disables all external consumers. Messages will only be processed using `InMemoryChannel`. Useful in testing environments. | `false`      |

***

## Buses

A **bus** defines how messages are routed through your system.

You can create multiple buses for different responsibilities, for example:

* command bus
* event bus
* query bus

#### Bus configuration

| Property   | Description                             | Default |
| ---------- | --------------------------------------- | ------- |
| `name`     | Unique name of the message bus.         | —       |
| `channels` | List of channel names used by this bus. | `[]`    |

Example:

```ts
buses: [
  {
    name: 'command.message-bus',
    channels: ['rabbitmq-channel'],
  },
  {
    name: 'event.message-bus',
    channels: ['rabbitmq-channel', 'in-memory-channel'],
  },
];
```

***

## Channels

Channels represent **transport layers** responsible for delivering and receiving messages.

Examples of transports:

* RabbitMQ
* Redis
* NATS
* Google Pub/Sub
* In-memory messaging

Each channel has its own configuration.

***

## InMemoryChannelConfig

The `InMemoryChannelConfig` is the simplest channel implementation and is useful for:

* development
* testing
* synchronous message execution

#### Properties

| Property                           | Description                                                           | Default            |
| ---------------------------------- | --------------------------------------------------------------------- | ------------------ |
| `name`                             | Name of the in-memory channel.                                        | —                  |
| `middlewares`                      | List of middlewares applied to messages passing through this channel. | `[]`               |
| `avoidErrorsForNotExistedHandlers` | Prevent errors when no handler exists for a message.                  | `false`            |
| `normalizer`                       | Custom message normalizer used before dispatching messages.           | Default normalizer |

Example:

```ts
channels: [
  new InMemoryChannelConfig({
    name: 'in-memory-channel',
    middlewares: [],
  }),
];
```


# Introduction

### @nestjstools/clock

**Time abstraction for NestJS**

In most applications, time is everywhere:

* timestamps
* expiration logic
* domain rules
* logging
* validations
* comparisons

But calling `new Date()` directly inside your business logic makes your system:

* Hard to test
* Non-deterministic
* Infrastructure-coupled
* Difficult to reason about

`@nestjstools/clock` solves this by introducing a clean abstraction over time.

### Why use Clock?

Instead of:

```
const now = new Date();
```

You write:

```
const now = this.clock.now();
```

And now:

* You can inject time
* You can freeze time in tests
* You remove infrastructure concerns from your domain layer

### The Problem with `new Date()`

Direct system time usage:

```
if (new Date() > expirationDate) { ... }
```

Creates:

* Hidden side effects
* Hard-to-test services
* Unpredictable unit tests
* Tight coupling to system clock

Time is infrastructure — it should be abstracted.

***

### The Solution: IClock

Clock introduces a simple interface:

```
interface IClock {
  now(): Date
  today(): CalendarDate
}
```

This allows you to:

* Inject system time
* Replace time in tests


# Installation

```
npm install @nestjstools/clock
```

or

```
yarn add @nestjstools/clock
```

## 🧩 NestJS Integration

Register the module:

```
import { Module } from '@nestjs/common';
import { ClockModule } from '@nestjstools/clock';

@Module({
  imports: [
    ClockModule.forRoot(), // global by default
  ],
})
export class AppModule {}
```

You can also use `.forFeature()` if needed.


# Usage example

### Injecting the Clock

```
import { Injectable } from '@nestjs/common';
import { IClock, Clock } from '@nestjstools/clock';

@Injectable()
export class SubscriptionService {
  constructor(@Clock() private readonly clock: IClock) {}

  isSubscriptionActive(startDate: Date, durationDays: number): boolean {
    const now = this.clock.now();

    const endDate = new Date(startDate);
    endDate.setDate(endDate.getDate() + durationDays);

    return now < endDate;
  }
}
```

Notice:

* No `new Date()` inside domain logic
* Time is injected
* Fully testable

***

## 🧪 Testing

Override the clock in tests:

```
import { Test } from '@nestjs/testing';
import { SubscriptionService } from './subscription.service';
import { FixedClock, Service } from '@nestjstools/clock';

describe('SubscriptionService', () => {
  let service: SubscriptionService;

  beforeEach(async () => {
    const moduleRef = await Test.createTestingModule({
      providers: [SubscriptionService],
    })
      .overrideProvider(Service.CLOCK_SERVICE)
      .useValue(new FixedClock(new Date('2020-10-10T00:00:00Z')))
      .compile();

    service = moduleRef.get(SubscriptionService);
  });

  it('returns true when subscription is active', () => {
    const start = new Date('2020-10-01T00:00:00Z');
    const active = service.isSubscriptionActive(start, 20);

    expect(active).toBe(true);
  });

  it('returns false when subscription expired', () => {
    const start = new Date('2020-09-01T00:00:00Z');
    const active = service.isSubscriptionActive(start, 20);

    expect(active).toBe(false);
  });
});
```

Deterministic. Predictable. Clean.


# Calendar value object

## 📅 CalendarDate Value Object

`CalendarDate` is an immutable date-only value object.

It represents a calendar date **without time and timezone**.

Perfect for:

* Birthdays
* Subscription periods
* Business deadlines
* Billing cycles
* Domain rules where time is irrelevant

***

### Features

* Immutable YYYY-MM-DD representation
* Creation from string or native Date
* Validation of invalid dates
* Safe add/subtract days
* Comparison helpers
* Conversion to native Date (00:00:00 time)

***

### Usage

```
import { CalendarDate } from '@nestjstools/clock';

// Create from string
const date1 = CalendarDate.fromString('2025-06-14');

// Create from native Date
const date2 = CalendarDate.fromDate(new Date());

// Get today
const today = CalendarDate.today();

// Manipulation
const nextWeek = today.addDays(7);
const yesterday = today.subtractDays(1);

// Comparison
if (date1.isBefore(nextWeek)) {
  console.log(`${date1.toString()} is before ${nextWeek.toString()}`);
}

// Convert to native Date
const nativeDate = date1.toDate();
```

***

## 🏗 Example: Using CalendarDate with IClock

```
import { Injectable } from '@nestjs/common';
import { IClock, Clock } from '@nestjstools/clock';

@Injectable()
export class ReturnToday {
  constructor(@Clock() private readonly clock: IClock) {}

  todayIs(): string {
    const today = this.clock.today();
    return today.toString(); // YYYY-MM-DD
  }
}
```


# Introduction

### @nestjstools/apisix-client

**Strongly-typed NestJS client for the Apache APISIX Admin API.**

`@nestjstools/apisix-client` allows you to programmatically manage:

* Routes
* Services
* Upstreams
* Consumers
* Plugins

directly from your NestJS application.

It is designed for:

* Infrastructure automation
* Platform tooling
* Gateway synchronization
* Dynamic multi-tenant systems
* Internal developer platforms

***

### What is Apache APISIX?

Apache APISIX is a high-performance, dynamic API Gateway built on OpenResty and etcd.

It provides:

* Routing
* Authentication
* Rate limiting
* Traffic shaping
* Plugin system
* Service discovery
* TLS management

But managing APISIX manually via Admin API is not scalable for modern platform teams.

***

### What This Client Solves

Instead of:

* Writing curl scripts
* Manually configuring routes
* Maintaining external provisioning scripts

You manage gateway configuration inside your NestJS platform services.

### Why This Library?

* Strong TypeScript typing
* Clean NestJS integration
* Declarative infrastructure approach
* Programmatic gateway management
* Automation-ready APIs


# Installation

This library requires `@nestjs/axios`.

```
npm install @nestjstools/apisix-client @nestjs/axios
```

or

```
yarn add @nestjstools/apisix-client @nestjs/axios
```

## Quick Start

### Register the Module

```
import { Module } from '@nestjs/common';
import { ApisixClientModule } from '@nestjstools/apisix-client';

@Module({
  imports: [
    ApisixClientModule.forRoot({
      url: 'http://localhost',
      adminSecret: process.env.APISIX_ADMIN_SECRET,
      global: true,      // optional (default: true)
      prefix: 'apisix',  // optional (default: 'apisix')
      port: 9180,        // optional (default: 9180)
    }),
  ],
})
export class AppModule {}
```

#### Base URL constructed as:

```
http://localhost:9180/apisix
```


# Usage example

### Inject the Client

```
import { Injectable, OnModuleInit } from '@nestjs/common';
import { ApisixClient, ApisixRouteRequest } from '@nestjstools/apisix-client';

@Injectable()
export class GatewaySyncService implements OnModuleInit {
  constructor(private readonly apisix: ApisixClient) {}

  async onModuleInit() {
    await this.ensureUserRoute();
  }

  private async ensureUserRoute() {
    const routeId = 'users-route';

    const desiredRoute: ApisixRouteRequest = {
      id: routeId,
      uri: '/users',
      methods: ['GET', 'POST'],
      upstream: {
        type: 'roundrobin',
        nodes: {
          'host.docker.internal:3000': 1,
        },
      },
    };

    await this.apisix.route().upsertRoute(routeId, desiredRoute);

    console.log('APISIX route updated');
  }
}
```

***

## 🔧 Configuration

| Property      | Description              | Default  |
| ------------- | ------------------------ | -------- |
| `url`         | Base APISIX host         | required |
| `adminSecret` | APISIX Admin API key     | required |
| `port`        | Admin API port           | `9180`   |
| `prefix`      | Admin API prefix         | `apisix` |
| `global`      | Register module globally | `true`   |

***

## 🛠 Supported Gateway Resources

The client supports management of:

* Routes
* Services
* Upstreams
* Consumers
* Plugins

Including commonly used plugin categories:

* Authentication (JWT, key-auth, etc.)
* Rate limiting
* Traffic control
* Transformation
* Security
* CORS
* Request validation


# Introduction

### @nestjstools/domain-driven-starter

**Framework-agnostic Domain-Driven Design primitives for TypeScript.**

This library provides foundational building blocks for implementing rich domain models using Domain-Driven Design (DDD) principles.

### What’s Included

* Immutable `Uuid` value object (v7, v4 compatible)
* `DomainEvent` interface
* Generic `AggregateRoot<T>` base class
* Event recording pattern
* Clean factory method enforcement
* Zero framework coupling

It is:

* Lightweight
* Infrastructure-free
* Fully testable
* Framework-agnostic
* Compatible with NestJS, Express, or standalone Node.js services


# Installation

```
npm install @nestjstools/domain-driven-starter
```

or

```
yarn add @nestjstools/domain-driven-starter
```


# Usage example

## Uuid Value Object

A strongly-typed, immutable UUID wrapper.

Prevents primitive obsession and ensures valid IDs.

```
import { Uuid } from '@nestjstools/domain-driven-starter';

const id = Uuid.generate();
const fromString = Uuid.fromString('f47ac10b-58cc-4372-a567-0e02b2c3d479');

console.log(id.toString());
```

#### Why use Uuid as a Value Object?

Instead of:

```
const id: string = '...';
```

You get:

* Validation
* Explicit domain meaning
* Type safety
* Immutable identity

Supports:

* UUID v7 (recommended)
* UUID v4 (compatible)

***

## 2️⃣ DomainEvent Interface

Minimal interface for domain events.

```
export interface DomainEvent {
  readonly id: string;
}
```

You define your own events:

```
export class OrderCreatedEvent implements DomainEvent {
  readonly occurredAt = new Date();
  readonly eventName = 'order.created';

  constructor(
    public readonly id: string,
    public readonly customerId: string
  ) {}
}
```

***

### Why Domain Events?

They allow:

* Decoupled side effects
* Clear domain traceability
* Event-driven architecture
* Integration with messaging systems
* Auditability

***

## 3️⃣ AggregateRoot\<T extends DomainEvent>

The central building block of your domain model.

Provides:

* Identity handling
* Event recording
* Controlled creation
* Event extraction

***

### Constructor Pattern

```
protected constructor(id: Uuid)
```

The constructor is **protected**, meaning:

* Only subclasses can instantiate it
* Prevents uncontrolled aggregate creation
* Encourages static factory methods

In your aggregate, you usually make the constructor `private` and expose a `createNew()` method.

***

### Methods

#### recordEvent(event: T): void

Records a domain event internally.

Used inside aggregate methods when something meaningful changes.

***

#### popRecordedEvents(): T\[]

Returns recorded events and clears the internal list.

Used after persisting the aggregate to publish events externally.

***

## 🧩 Example: Order Aggregate

```
import { AggregateRoot, Uuid } from '@nestjstools/domain-driven-starter';

export class OrderAggregate extends AggregateRoot<OrderEvents> {
  private constructor(
    id: Uuid,
    private readonly customerId: Uuid
  ) {
    super(id);
  }

  static createNew(
    id: Uuid,
    customerId: Uuid,
    now: Date
  ): OrderAggregate {
    const order = new OrderAggregate(id, customerId);

    order.recordEvent(
      new OrderCreatedEvent(
        id.toString(),
        customerId.toString()
      )
    );

    return order;
  }
}
```

***

## 🔁 Typical Usage Flow

1. Create aggregate via static factory.
2. Aggregate records domain events.
3. Repository saves aggregate.
4. Application layer calls `popRecordedEvents()`.
5. Events are published to an event bus (e.g. messaging library).

This keeps:

* Domain pure
* Infrastructure separate
* Side effects decoupled

***

## 🏗 Where This Library Fits

Perfect for:

* DDD projects
* CQRS systems
* Event-driven architecture
* Clean architecture
* Microservices
* Monoliths with strong domain boundaries

Works with:

* NestJS
* Express
* Fastify
* Standalone Node.js
* Any framework

***

## 🧪 Testing

Because it is framework-agnostic:

* No DI required
* No mocking frameworks needed
* Pure unit tests
* Aggregates are deterministic

Example:

```
it('records OrderCreatedEvent', () => {
  const id = Uuid.generate();
  const customerId = Uuid.generate();

  const order = OrderAggregate.createNew(id, customerId, new Date());

  const events = order.popRecordedEvents();

  expect(events).toHaveLength(1);
});
```