KubeMQ
IntegrationsNestJSHow-to guides

Set Up the CQRS Bridge

Install and configure the NestJS CQRS bridge to route @nestjs/cqrs commands, queries, and events through KubeMQ.

Wire up @kubemq/nestjs-transport/cqrs to distribute @nestjs/cqrs commands, queries, and events across services. For how the routing works internally, see CQRS Bridge Concepts; for every configuration field, see the KubeMQCqrsOptions reference.

Prerequisites

  • @kubemq/nestjs-transport and kubemq-js installed, with KubeMQModule.forRoot() already registered in the app (see Getting Started with NestJS)
  • The @nestjs/cqrs peer dependency installed (see Setup below)
  • A running KubeMQ broker reachable on localhost:50000

Setup

Install the optional peer dependency alongside the transport:

npm install @nestjs/cqrs

Import order matters. KubeMQCqrsModule.forRoot() swaps the bus publishers during its onModuleInit, so both CqrsModule (which provides the buses) and KubeMQModule.forRoot() (which provides the broker connection) must be initialized first. Import them before KubeMQCqrsModule.forRoot():

app.module.ts
import { Module } from '@nestjs/common';
import { CqrsModule } from '@nestjs/cqrs';
import { KubeMQModule } from '@kubemq/nestjs-transport';
import { KubeMQCqrsModule } from '@kubemq/nestjs-transport/cqrs';

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

@Module({
  imports: [
    KubeMQModule.forRoot({
      address,
      clientId: 'orders-app',
      isGlobal: true,
    }),
    CqrsModule,
    KubeMQCqrsModule.forRoot({
      commandChannelPrefix: 'myapp.commands',
      queryChannelPrefix: 'myapp.queries',
      eventChannelPrefix: 'myapp.events',
      commandTimeout: 10,
      queryTimeout: 10,
    }),
  ],
})
export class AppModule {}

KubeMQCqrsModule reads the broker address, credentials, TLS, and reconnection settings from the KubeMQModule.forRoot() registration — it does not take its own address. If KubeMQModule.forRoot() (or forRootAsync()) is missing, the module throws on startup because no connection configuration is available.

If CQRS commands, queries, or events are not being distributed, the most common cause is import order. KubeMQCqrsModule.forRoot() must be imported after both CqrsModule and KubeMQModule.forRoot(), otherwise the buses are not yet available to have their publishers swapped.

You need a running KubeMQ broker. The transport (and therefore the bridge) connects over native gRPC on port 50000 — no HTTP connector is required:

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

For async configuration (for example, pulling prefixes from ConfigService), use KubeMQCqrsModule.forRootAsync({ imports, useFactory, inject }), which returns the same providers via a factory. See the KubeMQCqrsOptions reference for the full options table.

Commands

A command is a plain class. Define it, annotate a handler with the @nestjs/cqrs @CommandHandler decorator, and dispatch it with CommandBus.execute(). The bridge serializes the command instance and sends it over the KubeMQ command channel; the handler's return value is sent back as the command response.

create-order.command.ts
export class CreateOrderCommand {
  constructor(
    public readonly productId: string,
    public readonly quantity: number,
  ) {}
}
create-order.handler.ts
import { CommandHandler, ICommandHandler, EventBus } from '@nestjs/cqrs';
import { Logger } from '@nestjs/common';
import { CreateOrderCommand } from './create-order.command';
import { OrderCreatedEvent } from './order-created.event';

@CommandHandler(CreateOrderCommand)
export class CreateOrderHandler implements ICommandHandler<CreateOrderCommand> {
  private readonly logger = new Logger(CreateOrderHandler.name);
  private orderCounter = 0;

  constructor(private readonly eventBus: EventBus) {}

  async execute(command: CreateOrderCommand): Promise<string> {
    this.orderCounter++;
    const orderId = `ORD-${String(this.orderCounter).padStart(3, '0')}`;
    this.logger.log(`Creating order ${orderId}: product=${command.productId}, qty=${command.quantity}`);

    this.eventBus.publish(new OrderCreatedEvent(orderId, command.productId, command.quantity));
    return orderId;
  }
}

Dispatch from any service that injects the CommandBus:

order.service.ts
import { Injectable } from '@nestjs/common';
import { CommandBus } from '@nestjs/cqrs';
import { CreateOrderCommand } from './create-order.command';

@Injectable()
export class OrderService {
  constructor(private readonly commandBus: CommandBus) {}

  async createOrder(productId: string, quantity: number): Promise<string> {
    // Routed to: myapp.commands.CreateOrderCommand
    return this.commandBus.execute<CreateOrderCommand, string>(
      new CreateOrderCommand(productId, quantity),
    );
  }
}

With the default commandChannelPrefix of cqrs.commands, this command travels to cqrs.commands.CreateOrderCommand. If the handler reports a non-executed response, the bridge throws an error that includes the failing channel name.

Queries

Queries work like commands but use the QueryBus and the @QueryHandler decorator. The query is routed to a query channel, handled remotely, and the result is returned to the caller.

get-order.query.ts
export class GetOrderQuery {
  constructor(public readonly orderId: string) {}
}

export class GetOrderResult {
  constructor(
    public readonly orderId: string,
    public readonly productId: string,
    public readonly quantity: number,
    public readonly status: string,
  ) {}
}
get-order.handler.ts
import { QueryHandler, IQueryHandler } from '@nestjs/cqrs';
import { Logger } from '@nestjs/common';
import { GetOrderQuery, GetOrderResult } from './get-order.query';
import { OrderStore } from './order.store';

@QueryHandler(GetOrderQuery)
export class GetOrderHandler implements IQueryHandler<GetOrderQuery, GetOrderResult> {
  private readonly logger = new Logger(GetOrderHandler.name);

  constructor(private readonly store: OrderStore) {}

  async execute(query: GetOrderQuery): Promise<GetOrderResult> {
    this.logger.log(`Querying order: ${query.orderId}`);
    const order = this.store.findById(query.orderId);
    if (!order) {
      throw new Error(`Order ${query.orderId} not found`);
    }
    return new GetOrderResult(order.orderId, order.productId, order.quantity, order.status);
  }
}
order.service.ts
import { Injectable } from '@nestjs/common';
import { QueryBus } from '@nestjs/cqrs';
import { GetOrderQuery, GetOrderResult } from './get-order.query';

@Injectable()
export class OrderQueryService {
  constructor(private readonly queryBus: QueryBus) {}

  async getOrder(orderId: string): Promise<GetOrderResult> {
    // Routed to: myapp.queries.GetOrderQuery
    return this.queryBus.execute<GetOrderQuery, GetOrderResult>(
      new GetOrderQuery(orderId),
    );
  }
}

The QueryBus resolves with the deserialized response body, so the caller awaits the result exactly as it would with the in-process @nestjs/cqrs bus.

Events

Events are dispatched through the EventBus and are fire-and-forget: EventBus.publish() fans the event out over KubeMQ to every subscribed handler. Use the @nestjs/cqrs @EventsHandler decorator on the receiving side.

order-created.event.ts
export class OrderCreatedEvent {
  constructor(
    public readonly orderId: string,
    public readonly productId: string,
    public readonly quantity: number,
  ) {}
}
order-created.handler.ts
import { EventsHandler, IEventHandler } from '@nestjs/cqrs';
import { Logger } from '@nestjs/common';
import { OrderCreatedEvent } from './order-created.event';
import { OrderStore } from './order.store';

@EventsHandler(OrderCreatedEvent)
export class OrderCreatedHandler implements IEventHandler<OrderCreatedEvent> {
  private readonly logger = new Logger(OrderCreatedHandler.name);

  constructor(private readonly store: OrderStore) {}

  handle(event: OrderCreatedEvent): void {
    this.logger.log(`Event received — storing order ${event.orderId}`);
    this.store.save({
      orderId: event.orderId,
      productId: event.productId,
      quantity: event.quantity,
      status: 'created',
    });
  }
}
order.service.ts
import { Injectable, Logger } from '@nestjs/common';
import { EventBus } from '@nestjs/cqrs';
import { OrderCreatedEvent } from './order-created.event';

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

  constructor(private readonly eventBus: EventBus) {}

  async createOrder(orderId: string, product: string, quantity: number): Promise<void> {
    this.logger.log(`Publishing OrderCreatedEvent: id=${orderId}`);
    // Routed to: myapp.events.OrderCreatedEvent
    this.eventBus.publish(new OrderCreatedEvent(orderId, product, quantity));
    this.logger.log('Event published successfully');
  }
}

By default events use the transient Events pattern (sendEvent) — if no handler is subscribed when the event is published, it is not retained. For durable, replayable domain events, set persistEvents: true so the bridge publishes to the KubeMQ Events Store (sendEventStore) instead:

app.module.ts
KubeMQCqrsModule.forRoot({
  eventChannelPrefix: 'myapp.events',
  persistEvents: true,
})

Events Store retains the event stream so late subscribers can replay history. See Usage for the distinction between Events and Events Store, including start positions and sequence tracking.

Custom Channel Resolution

The channel segment after the prefix defaults to the message's constructor.name. So new CreateOrderCommand(...) lands on {commandChannelPrefix}.CreateOrderCommand. The default resolver falls back to unknown for anonymous or plain Object instances.

To map message classes onto your own channel naming scheme — for example versioned or namespaced channels — provide a channelResolver:

app.module.ts
KubeMQCqrsModule.forRoot({
  commandChannelPrefix: 'myapp.commands',
  // CreateOrderCommand -> myapp.commands.orders.create.v1
  channelResolver: (message) => {
    const map: Record<string, string> = {
      CreateOrderCommand: 'orders.create.v1',
      GetOrderQuery: 'orders.get.v1',
    };
    return map[message.constructor.name] ?? message.constructor.name;
  },
})

The same resolver is applied to commands, queries, and events, so make sure each message class resolves to a segment that is unique within its prefix.

End-to-End Example

The repository ships a complete full-cqrs-flow example that wires all three buses into a single command → event → query story: a CreateOrderCommand is dispatched, its handler publishes an OrderCreatedEvent that updates a read model, and a GetOrderQuery then reads that model back — every hop crossing KubeMQ.

app.module.ts
import { Module } from '@nestjs/common';
import { CqrsModule } from '@nestjs/cqrs';
import { KubeMQModule } from '@kubemq/nestjs-transport';
import { KubeMQCqrsModule } from '@kubemq/nestjs-transport/cqrs';
import { CreateOrderHandler } from './create-order.handler';
import { OrderCreatedHandler } from './order-created.handler';
import { GetOrderHandler } from './get-order.handler';
import { OrderStore } from './order.store';
import { OrderService } from './order.service';

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

@Module({
  imports: [
    KubeMQModule.forRoot({
      address,
      clientId: 'nestjs-cqrs-full-flow',
      isGlobal: true,
    }),
    CqrsModule,
    KubeMQCqrsModule.forRoot({
      commandChannelPrefix: 'nestjs-cqrs.full-flow.commands',
      queryChannelPrefix: 'nestjs-cqrs.full-flow.queries',
      eventChannelPrefix: 'nestjs-cqrs.full-flow.events',
      commandTimeout: 10,
      queryTimeout: 10,
    }),
  ],
  providers: [
    OrderStore,
    CreateOrderHandler,
    OrderCreatedHandler,
    GetOrderHandler,
    OrderService,
  ],
})
export class AppModule {}

Run it against a live broker:

cd examples
npm install
npx tsx examples/cqrs/full-cqrs-flow/main.ts

Expected output:

[FullCqrsFlowExample] KubeMQ microservice started
[OrderService] Step 1: Dispatching CreateOrderCommand for product=WIDGET-42
[CreateOrderHandler] Creating order ORD-001: product=WIDGET-42, qty=3
[OrderCreatedHandler] Event received — storing order ORD-001
[OrderService] Step 1 complete: orderId=ORD-001
[OrderService] Step 2: Querying order ORD-001
[GetOrderHandler] Querying order: ORD-001
[OrderService] Step 2 complete: {"orderId":"ORD-001","productId":"WIDGET-42","quantity":3,"status":"created"}

Focused single-bus examples are available too: examples/cqrs/cqrs-commands, examples/cqrs/cqrs-queries, and examples/cqrs/cqrs-events. Each example reads the broker address from KUBEMQ_ADDRESS and falls back to localhost:50000.

CQRS Bridge Concepts

How the bridge routes CommandBus, QueryBus, and EventBus traffic over KubeMQ.

KubeMQCqrsOptions reference

Every KubeMQCqrsModule.forRoot() configuration field.

Usage

Commands, Queries, Events, Events Store, and Queues with the handler decorators and KubeMQRecord builder.

Was this page helpful?

Configuration & Resilience

Configure TLS/auth, reconnection, custom serialization, DLQ routing, validation, idempotency, and a circuit breaker for production handlers.

Module Configuration

Wire the transport into NestJS DI with forRoot, register, async factories, multi-broker, and forFeature scoped clients.

On this page

PrerequisitesSetupCommandsQueriesEventsCustom Channel ResolutionEnd-to-End ExampleRelated