The Transport Model

@kubemq/nestjs-transport plugs KubeMQ into the NestJS microservices abstraction. NestJS splits a microservice into two halves — an inbound strategy that receives messages and dispatches them to handlers, and an outbound client proxy that sends messages out. The transport implements both halves against KubeMQ:

KubeMQServer implements the NestJS CustomTransportStrategy interface. You pass an instance to app.connectMicroservice({ strategy }), and it subscribes to KubeMQ channels and routes incoming messages to your decorated handlers.
KubeMQClientProxy extends the NestJS ClientProxy base class. You register it through the module system and inject it as a standard ClientProxy, then call send() and emit() to publish messages.

Both halves wrap a kubemq-js client, which speaks KubeMQ's native gRPC protocol on port 50000. No HTTP connector is involved — the transport talks to the broker directly.

main.ts

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

async function bootstrap() {
  const app = await NestFactory.create(AppModule);

  // KubeMQServer is the inbound CustomTransportStrategy
  app.connectMicroservice({
    strategy: new KubeMQServer({
      address: 'localhost:50000',
      clientId: 'my-server',
      group: 'my-group',
    }),
  });

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

A local broker for development needs only the native gRPC port:

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

Port 50000 is the gRPC port that native SDKs and this transport use. If you also run the HTTP connectors (REST, CloudEvents, MCP, A2A) you would expose the shared HTTP server on port 9090, but the NestJS transport does not require it.

`send()` vs `emit()` Mapping

NestJS gives ClientProxy two methods, and the transport assigns each a default KubeMQ pattern:

client.send(channel, data) is request-reply. By default it sends a Command and waits for the handler's return value to come back over a reply channel.
client.emit(channel, data) is fire-and-forget. By default it sends an Event with no response.

When you need a different pattern, wrap the payload in a KubeMQRecord and call a builder method to re-target the message type. The builder does not change which method you call — send() stays request-reply, emit() stays fire-and-forget — it changes the KubeMQ pattern the message is delivered as.

Builder	Pattern type	Client API
(default)	Command	`client.send(channel, data)`
`.asQuery()`	Query	`client.send(channel, record)`
(default)	Event	`client.emit(channel, data)`
`.asEventStore()`	Events Store	`client.emit(channel, record)`
`.asQueue()`	Queue	`client.emit(channel, record)`

order.service.ts

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

@Injectable()
export class OrderService {
  constructor(@Inject('ORDER_SERVICE') private client: ClientProxy) {}

  // Command (default) — request-reply
  createOrder(data: { name: string; total: number }) {
    return firstValueFrom(this.client.send('orders.create', data));
  }

  // Query — request-reply, re-targeted with .asQuery()
  getOrder(id: string) {
    const record = new KubeMQRecord({ id }).asQuery();
    return firstValueFrom(this.client.send('orders.get', record));
  }

  // Event (default) — fire-and-forget
  notifyUpdated(id: string, status: string) {
    return firstValueFrom(this.client.emit('orders.updated', { id, status }));
  }

  // Queue — fire-and-forget, re-targeted with .asQueue()
  enqueueProcessing(id: string) {
    const record = new KubeMQRecord({ id }).asQueue();
    return firstValueFrom(this.client.emit('orders.process', record));
  }
}

The Five Patterns at a Glance

KubeMQ exposes five messaging patterns. The transport surfaces each one as a custom handler decorator. Under the hood, request-reply patterns build on the NestJS @MessagePattern and fire-and-forget patterns build on @EventPattern — the custom decorators just attach the right KubeMQ pattern metadata so you never configure it by hand.

Decorator	NestJS equivalent	Pattern type	Direction
`@CommandHandler`	`@MessagePattern(ch, { type: 'command' })`	Command	Request-reply
`@QueryHandler`	`@MessagePattern(ch, { type: 'query' })`	Query	Request-reply
`@EventHandler`	`@EventPattern(ch, { type: 'event' })`	Event	Fire-and-forget
`@EventStoreHandler`	`@EventPattern(ch, { type: 'event_store' })`	Events Store	Fire-and-forget
`@QueueHandler`	`@EventPattern(ch, { type: 'queue' })`	Queue	Fire-and-forget

order.handler.ts

import {
  CommandHandler,
  QueryHandler,
  EventHandler,
  EventStoreHandler,
  QueueHandler,
  KubeMQCommandContext,
  KubeMQQueryContext,
  KubeMQContext,
  KubeMQEventStoreContext,
  KubeMQQueueContext,
} from '@kubemq/nestjs-transport';
import { Payload, Ctx } from '@nestjs/microservices';

export class OrderHandler {
  @CommandHandler('orders.create')
  create(@Payload() data: { name: string }, @Ctx() ctx: KubeMQCommandContext) {
    return { orderId: 'order-123', status: 'created' };
  }

  @QueryHandler('orders.get')
  get(@Payload() data: { id: string }, @Ctx() ctx: KubeMQQueryContext) {
    return { orderId: data.id, name: 'Widget' };
  }

  @EventHandler('orders.updated')
  updated(@Payload() data: { id: string }, @Ctx() ctx: KubeMQContext) {
    // no return value — fire-and-forget
  }

  @EventStoreHandler('orders.history', { startFrom: 'first' })
  history(@Payload() data: unknown, @Ctx() ctx: KubeMQEventStoreContext) {
    // ctx.sequence is the stored event position
  }

  @QueueHandler('orders.process', { maxMessages: 1 })
  process(@Payload() data: unknown, @Ctx() ctx: KubeMQQueueContext) {
    // ctx.receiveCount tracks delivery attempts
  }
}

Message Tags & Metadata

Every message the transport sends carries a small set of KubeMQ tags that let the receiving KubeMQServer reconstruct the routing context. The pattern, type, message id, and content type travel as the following keys (exported from src/constants.ts):

Constant	Tag key	Purpose
`TAG_PATTERN`	`nestjs:pattern`	The original NestJS pattern (channel) string
`TAG_ID`	`nestjs:id`	Unique message id
`TAG_TYPE`	`nestjs:type`	KubeMQ pattern type (`command`, `query`, `event`, `event_store`, `queue`)
`TAG_CONTENT_TYPE`	`nestjs:content-type`	Serializer content type hint

Advanced features carry additional well-known tags that follow industry conventions, so they interoperate with non-NestJS producers and tracing tools:

Constant	Tag key	Purpose
`TAG_CORRELATION_ID`	`x-correlation-id`	Correlation id for a logical flow
`TAG_CAUSATION_ID`	`x-causation-id`	Id of the message that caused this one
`TAG_IDEMPOTENCY_KEY`	`x-idempotency-key`	De-duplication key
`TAG_TRACEPARENT`	`traceparent`	W3C Trace Context parent
`TAG_TRACESTATE`	`tracestate`	W3C Trace Context vendor state

Correlation and causation flow through a CorrelationContext backed by Node's AsyncLocalStorage. When a message arrives, the transport reads the x-correlation-id tag (or generates a new UUID if absent) and sets the incoming message's id as the causation id, so any messages your handler emits are automatically linked back to it:

correlation-context.ts (excerpt)

static createFromTags(
  tags: Record<string, string> | undefined,
  messageId: string,
  correlationIdTag: string,
  _causationIdTag: string,
): CorrelationStore {
  const correlationId = tags?.[correlationIdTag] ?? randomUUID();
  const causationId = messageId;
  return { correlationId, causationId };
}

Inside a handler you can read these directly off the context:

@CommandHandler('orders.create')
create(@Payload() data: unknown, @Ctx() ctx: KubeMQCommandContext) {
  const correlationId = ctx.getCorrelationId(); // from x-correlation-id tag
  const causationId = ctx.getCausationId();     // from x-causation-id tag
  // ...
}

Context Hierarchy

Every handler receives a context object via the NestJS @Ctx() parameter decorator. All contexts derive from KubeMQContext, which extends the NestJS BaseRpcContext. Each pattern then adds the fields that are meaningful for it.

The base KubeMQContext exposes the shared routing data:

kubemq.context.ts (excerpt)

export class KubeMQContext extends BaseRpcContext<[Record<string, any>]> {
  get channel(): string { /* ... */ }
  get id(): string { /* ... */ }
  get timestamp(): Date { /* ... */ }
  get tags(): Record<string, string> { /* ... */ }
  get metadata(): string { /* ... */ }
  get patternType(): KubeMQPatternType { /* ... */ }

  getCorrelationId(): string | undefined { /* ... */ }
  getCausationId(): string | undefined { /* ... */ }
}

The pattern-specific contexts add to that base:

Context	Used by	Adds
`KubeMQContext`	`@EventHandler`	`channel`, `id`, `timestamp`, `tags`, `metadata`, `patternType`
`KubeMQCommandContext`	`@CommandHandler`	`fromClientId`, `replyChannel`
`KubeMQQueryContext`	`@QueryHandler`	`fromClientId`, `replyChannel`
`KubeMQEventStoreContext`	`@EventStoreHandler`	`sequence`
`KubeMQQueueContext`	`@QueueHandler`	`sequence`, `receiveCount`, `isReRouted`, `ack()`, `nack()`, `reQueue(channel)`

The command and query contexts surface fromClientId (the caller's client id) and replyChannel (where the response is routed) because request-reply needs to know who is waiting. The events store context adds the stored sequence so replay handlers know their position. The queue context is the richest — it adds delivery bookkeeping and, in manual-ack mode, the methods that settle a message:

kubemq-queue.context.ts (excerpt)

export class KubeMQQueueContext extends KubeMQContext {
  get sequence(): number { /* ... */ }
  get receiveCount(): number { /* ... */ }
  get isReRouted(): boolean { /* ... */ }

  ack(): void { /* requires { manualAck: true } */ }
  nack(): void { /* requires { manualAck: true } */ }
  reQueue(channel: string): void { /* requires { manualAck: true } */ }
}

order.handler.ts (manual ack)

@QueueHandler('orders.process', { manualAck: true })
async process(@Payload() data: unknown, @Ctx() ctx: KubeMQQueueContext) {
  try {
    await processOrder(data);
    ctx.ack();
  } catch {
    ctx.reQueue('orders.dlq');
  }
}

ack(), nack(), and reQueue() are only available when the handler is declared with { manualAck: true }. Calling them in auto-ack mode throws an error.

Serialization Pipeline

Message payloads cross the wire as bytes. On the way out the KubeMQSerializer turns your object into a Uint8Array; on the way in the KubeMQDeserializer turns those bytes back into an object. Both default to JSON (JsonSerializer / JsonDeserializer).

serialization/interfaces.ts

export interface KubeMQSerializer {
  serialize(value: unknown): Uint8Array;
  readonly contentType?: string;
}

export interface KubeMQDeserializer {
  deserialize(data: Uint8Array, tags?: Record<string, string>): unknown;
}

The serializer and deserializer must match on both the server and the client. If a client serializes with MessagePack but the server tries to deserialize as JSON, a SerializationError is thrown. Configure the pair identically on the KubeMQServer strategy and on every KubeMQModule.register() client:

import { MessagePackSerializer, MessagePackDeserializer } from '@kubemq/nestjs-transport';

// Server
new KubeMQServer({
  address: 'localhost:50000',
  serializer: new MessagePackSerializer(),
  deserializer: new MessagePackDeserializer(),
});

// Client
KubeMQModule.register({
  name: 'ORDER_SERVICE',
  address: 'localhost:50000',
  serializer: new MessagePackSerializer(),
  deserializer: new MessagePackDeserializer(),
});

A serialization mismatch — different serializer/deserializer pairs on the two ends — surfaces as a SerializationError when deserialization fails. Always keep the codec configuration symmetric across sender and receiver.

The transport ships JSON (default), MessagePack (MessagePackSerializer / MessagePackDeserializer), and Protobuf (ProtobufSerializer / ProtobufDeserializer) codecs, and you can supply your own by implementing the two interfaces.

Connection Lifecycle

The underlying kubemq-js client manages a long-lived gRPC connection whose state is reported as a KubeMQStatus enum:

events/kubemq.events.ts

export const enum KubeMQStatus {
  DISCONNECTED = 'disconnected',
  RECONNECTING = 'reconnecting',
  CONNECTED = 'connected',
  CLOSED = 'closed',
}

Two mechanisms keep the transport resilient:

Startup gating with waitForConnection (default true) — the KubeMQServer blocks startAllMicroservices() until it reaches CONNECTED, so your service does not begin handling traffic against an unready broker. Set it to false for non-blocking startup that connects in the background.
Automatic reconnection with ReconnectionPolicy — when the connection drops, the client moves to RECONNECTING and retries with exponential backoff and jitter, then returns to CONNECTED once the broker is reachable again.

new KubeMQServer({
  address: 'localhost:50000',
  waitForConnection: true,
  reconnect: {
    maxAttempts: -1,        // unlimited
    initialDelayMs: 500,
    maxDelayMs: 30_000,
    multiplier: 2.0,
    jitter: 'full',
  },
});

Parameter	Default	Description
`maxAttempts`	`-1`	Maximum reconnection attempts (`-1` = unlimited)
`initialDelayMs`	`500`	Initial backoff interval (milliseconds)
`maxDelayMs`	`30000`	Maximum backoff interval (milliseconds)
`multiplier`	`2.0`	Backoff multiplier
`jitter`	`'full'`	Jitter strategy (`'none'` \| `'full'` \| `'decorrelated'`)

Command Round-Trip

Putting the pieces together, here is how a single command flows from a service call to a handler and back. The client send() produces a Command, the broker routes it to the subscribed KubeMQServer, the @CommandHandler runs, and its return value travels back over the reply channel recorded in KubeMQCommandContext.

NestJS Transport Concepts

The Transport Model

`send()` vs `emit()` Mapping

The Five Patterns at a Glance

Message Tags & Metadata

Context Hierarchy

Serialization Pipeline

Connection Lifecycle

Command Round-Trip

Events

Events Store

Queues

RPC

On this page