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.

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.

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.

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.

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.

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.

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.

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.

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

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