Overview

The KubeMQ transport plugs into NestJS dependency injection through the KubeMQModule dynamic module. It exposes two distinct roles that you compose in your AppModule:

forRoot / forRootAsync register the global server/connection configuration that the transport shares across the application. They provide the KUBEMQ_MODULE_OPTIONS token and a shared KubeMQClientProxy under KUBEMQ_SDK_CLIENT.
register / registerAsync create a named KubeMQClientProxy that you inject into your services via @Inject(name). This is the client you call send() and emit() on.

A third role, forFeature / forFeatureAsync, derives a scoped client from the global connection — every channel gets a channelPrefix automatically prepended.

Two separate things are configured here. The module (KubeMQModule) wires up DI and outbound clients. The inbound strategy (KubeMQServer) — which dispatches messages to your @*Handler methods — is constructed separately and passed to app.connectMicroservice(). See Connecting the microservice below.

Start a local broker before running any of the examples. The gRPC API the transport speaks to listens on 50000; the shared HTTP server (dashboard and connector endpoints) is on 9090:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

forRoot / forRootAsync

forRoot registers a global KubeMQ configuration shared by the whole application. Pass isGlobal: true so any module can resolve the shared client without re-importing KubeMQModule.

app.module.ts

import { Module } from '@nestjs/common';
import { KubeMQModule } from '@kubemq/nestjs-transport';

const address = process.env.KUBEMQ_ADDRESS ?? 'localhost:50000';

@Module({
  imports: [
    KubeMQModule.forRoot({
      address,
      clientId: 'nestjs-module-config-for-root-server',
      isGlobal: true,
      reconnect: {
        maxAttempts: -1,
        initialDelayMs: 1000,
        maxDelayMs: 30000,
        multiplier: 2,
        jitter: 'full',
      },
    }),
  ],
})
export class AppModule {}

When configuration must be resolved at runtime — from environment variables, a secrets manager, or any other async source — use forRootAsync with an imports / useFactory / inject triple. The factory returns KubeMQModuleOptions and receives whatever providers you list in inject:

app.module.ts

import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { KubeMQModule, type KubeMQModuleOptions } from '@kubemq/nestjs-transport';

@Module({
  imports: [
    ConfigModule.forRoot({ isGlobal: true }),
    KubeMQModule.forRootAsync({
      imports: [ConfigModule],
      useFactory: (config: ConfigService): KubeMQModuleOptions => ({
        address: config.get('KUBEMQ_ADDRESS', 'localhost:50000'),
        clientId: 'nestjs-module-config-for-root-async-server',
        reconnect: {
          maxAttempts: -1,
          initialDelayMs: 1000,
          maxDelayMs: 30000,
          multiplier: 2,
          jitter: 'full',
        },
      }),
      inject: [ConfigService],
    }),
  ],
})
export class AppModule {}

Instead of useFactory, you can supply a provider class. useClass implements KubeMQOptionsFactory (a class with a createKubeMQOptions() method) and is instantiated by Nest; useExisting reuses a factory provider that already exists in the DI container.

Parameters

forRoot accepts KubeMQModuleOptions (the full KubeMQServerOptions plus isGlobal). forRootAsync accepts the async configuration below.

Prop

Type

register / registerAsync

register creates a named KubeMQClientProxy and exports it under the token you choose. Give it a name, an address, and a clientId:

app.module.ts

import { Module } from '@nestjs/common';
import { KubeMQModule } from '@kubemq/nestjs-transport';
import { OrderHandlerService } from './order.handler.js';
import { OrderService } from './order.service.js';
import { SenderService } from './sender.service.js';

const address = process.env.KUBEMQ_ADDRESS ?? 'localhost:50000';

@Module({
  imports: [
    KubeMQModule.forRoot({
      address,
      clientId: 'nestjs-module-config-register-server',
      isGlobal: true,
    }),
    KubeMQModule.register({
      name: 'ORDER_SERVICE',
      address,
      clientId: 'nestjs-module-config-register-client',
    }),
  ],
  providers: [OrderHandlerService, OrderService, SenderService],
})
export class AppModule {}

Inject the named client into a service with @Inject('ORDER_SERVICE'). The proxy is a NestJS ClientProxy, so send() returns an Observable — wrap it in firstValueFrom to await a value:

sender.service.ts

import { Injectable, Inject, Logger } from '@nestjs/common';
import { KubeMQClientProxy } from '@kubemq/nestjs-transport';
import { firstValueFrom } from 'rxjs';

@Injectable()
export class SenderService {
  private readonly logger = new Logger(SenderService.name);

  constructor(@Inject('ORDER_SERVICE') private readonly client: KubeMQClientProxy) {}

  async sendEvent(): Promise<void> {
    await this.client.connect();
    this.logger.log('Sending event via named ORDER_SERVICE client...');
    await firstValueFrom(
      this.client.emit('nestjs-module-config.register', {
        orderId: 'ORD-100',
        product: 'Widget',
        quantity: 3,
      }),
    );
    this.logger.log('Event sent successfully');
  }
}

You can inject the client typed as the framework's ClientProxy from @nestjs/microservices or as the concrete KubeMQClientProxy. The latter gives you the connect() method shown above; the underlying token is the same.

registerAsync resolves client options at runtime, the same way forRootAsync does. The factory returns KubeMQClientOptions and pulls the address from ConfigService:

app.module.ts

import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import {
  KubeMQModule,
  type KubeMQModuleOptions,
  type KubeMQClientOptions,
} from '@kubemq/nestjs-transport';

@Module({
  imports: [
    ConfigModule.forRoot({ isGlobal: true }),
    KubeMQModule.forRootAsync({
      useFactory: (config: ConfigService): KubeMQModuleOptions => ({
        address: config.get('KUBEMQ_ADDRESS', 'localhost:50000'),
        clientId: 'nestjs-module-config-register-async-server',
      }),
      inject: [ConfigService],
    }),
    KubeMQModule.registerAsync({
      name: 'NOTIFICATION_SERVICE',
      useFactory: (config: ConfigService): KubeMQClientOptions => ({
        address: config.get('KUBEMQ_ADDRESS', 'localhost:50000'),
        clientId: 'nestjs-module-config-register-async-client',
      }),
      inject: [ConfigService],
    }),
  ],
})
export class AppModule {}

Like the root variants, registerAsync also accepts useClass / useExisting factories — here implementing KubeMQClientOptionsFactory.createKubeMQClientOptions().

Multi-broker

Because each register call produces an independent client bound to its own DI token, you can register multiple named clients against different addresses. This is the basis for fan-in / fan-out topologies that span more than one broker:

app.module.ts

import { Module } from '@nestjs/common';
import { KubeMQModule } from '@kubemq/nestjs-transport';
import { MultiBrokerHandlerService } from './multi-broker.handler.js';
import { MultiBrokerService } from './multi-broker.service.js';

const primaryAddress = process.env.KUBEMQ_PRIMARY_ADDRESS ?? 'localhost:50000';
const secondaryAddress = process.env.KUBEMQ_SECONDARY_ADDRESS ?? 'localhost:50000';

@Module({
  imports: [
    KubeMQModule.forRoot({
      address: primaryAddress,
      clientId: 'nestjs-module-config-multi-broker-server',
      isGlobal: true,
    }),
    KubeMQModule.register({
      name: 'PRIMARY_BROKER',
      address: primaryAddress,
      clientId: 'nestjs-module-config-multi-broker-primary',
    }),
    KubeMQModule.register({
      name: 'SECONDARY_BROKER',
      address: secondaryAddress,
      clientId: 'nestjs-module-config-multi-broker-secondary',
    }),
  ],
  providers: [MultiBrokerHandlerService, MultiBrokerService],
})
export class AppModule {}

A service then injects each broker by its own token and targets them independently:

multi-broker.service.ts

import { Injectable, Inject, Logger } from '@nestjs/common';
import { KubeMQClientProxy } from '@kubemq/nestjs-transport';
import { firstValueFrom } from 'rxjs';

@Injectable()
export class MultiBrokerService {
  private readonly logger = new Logger(MultiBrokerService.name);

  constructor(
    @Inject('PRIMARY_BROKER') private readonly primary: KubeMQClientProxy,
    @Inject('SECONDARY_BROKER') private readonly secondary: KubeMQClientProxy,
  ) {}

  async sendToPrimary(): Promise<void> {
    await this.primary.connect();
    await firstValueFrom(
      this.primary.emit('nestjs-module-config.multi-broker-primary', {
        source: 'primary',
        message: 'Hello from primary broker',
      }),
    );
  }

  async sendToSecondary(): Promise<void> {
    await this.secondary.connect();
    await firstValueFrom(
      this.secondary.emit('nestjs-module-config.multi-broker-secondary', {
        source: 'secondary',
        message: 'Hello from secondary broker',
      }),
    );
  }
}

The two addresses may point at the same broker with different client identities, or at genuinely separate clusters — the registration is identical either way.

forFeature / forFeatureAsync

forFeature creates a feature-scoped client that inherits its connection configuration from forRoot and prepends a channelPrefix to every send / emit pattern. It is backed by ScopedKubeMQClientProxy, which delegates to the shared client registered by forRoot (the KUBEMQ_SDK_CLIENT provider) and rewrites the channel name on each call.

orders.module.ts

import { Module } from '@nestjs/common';
import { KubeMQModule } from '@kubemq/nestjs-transport';
import { OrdersService } from './orders.service.js';

@Module({
  imports: [
    // Connection config inherited from the global forRoot() registration.
    KubeMQModule.forFeature({
      name: 'ORDERS_CLIENT',
      channelPrefix: 'orders.',
    }),
  ],
  providers: [OrdersService],
})
export class OrdersModule {}

With the prefix orders. in place, a service that emits to created actually targets orders.created, and send('get') targets orders.get — the prefix is applied transparently:

orders.service.ts

import { Injectable, Inject } from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';
import { firstValueFrom } from 'rxjs';

@Injectable()
export class OrdersService {
  constructor(@Inject('ORDERS_CLIENT') private readonly client: ClientProxy) {}

  async created(orderId: string): Promise<void> {
    // Emits to "orders.created"
    await firstValueFrom(this.client.emit('created', { orderId }));
  }
}

KubeMQFeatureOptions carries only the scoped concerns — the injection token (name) and the optional channelPrefix. There is no address or clientId here, because the connection is reused from forRoot. For runtime-resolved prefixes, forFeatureAsync accepts the familiar imports / useFactory / inject (and useClass / useExisting) shape, where the factory returns KubeMQFeatureOptions.

forFeature requires a prior forRoot (or forRootAsync) registration — the scoped proxy injects the shared KUBEMQ_SDK_CLIENT and will fail to resolve if no global connection exists.

Connecting the microservice (server side)

The module configures DI and outbound clients. It does not start the inbound listener. To dispatch incoming messages to your @CommandHandler / @EventHandler (and the other three) methods, construct a KubeMQServer strategy and pass it to app.connectMicroservice(), then call startAllMicroservices():

main.ts

import 'reflect-metadata';
import { NestFactory } from '@nestjs/core';
import { KubeMQServer } from '@kubemq/nestjs-transport';
import { AppModule } from './app.module.js';

async function bootstrap(): Promise<void> {
  const app = await NestFactory.create(AppModule);

  // Inbound: the strategy is constructed separately from the module
  // and dispatches messages to the @*Handler methods.
  app.connectMicroservice({
    strategy: new KubeMQServer({
      address: process.env.KUBEMQ_ADDRESS ?? 'localhost:50000',
      clientId: 'nestjs-module-config-for-root-server',
    }),
  });

  await app.startAllMicroservices();
  await app.listen(3000);
}
bootstrap();

Keep the distinction clear:

Concern	Where it lives	Role
`KubeMQModule.forRoot` / `register` / `forFeature`	`app.module.ts` imports	DI configuration and outbound clients (`KubeMQClientProxy`)
`new KubeMQServer({ ... })`	`main.ts`, passed to `connectMicroservice`	Inbound transport strategy that routes messages to handlers

The same instance you pass to connectMicroservice is also what a KubeMQHealthIndicator should wrap, so health checks share the transport's connection and subscription error map.

Module Configuration

Overview

forRoot / forRootAsync

Parameters

register / registerAsync

Multi-broker

forFeature / forFeatureAsync

Connecting the microservice (server side)

Common gotchas

Usage

Configuration reference

Getting Started

On this page

Module Configuration

Handler decorators do not fire

Messages are sent but never received

Scoped client cannot be resolved

Usage

Configuration reference

Getting Started

On this page