# NestJS Transport Concepts (/integrations/nestjs/concepts) ## The Transport Model [#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`](https://www.npmjs.com/package/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. ```typescript title="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: 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 [#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)` | ```typescript title="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 [#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 | ```typescript title="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 [#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: ```typescript title="correlation-context.ts (excerpt)" static createFromTags( tags: Record | 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: ```typescript @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 [#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: ```typescript title="kubemq.context.ts (excerpt)" export class KubeMQContext extends BaseRpcContext<[Record]> { get channel(): string { /* ... */ } get id(): string { /* ... */ } get timestamp(): Date { /* ... */ } get tags(): Record { /* ... */ } 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: ```typescript title="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 } */ } } ``` ```typescript title="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 [#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`). ```typescript title="serialization/interfaces.ts" export interface KubeMQSerializer { serialize(value: unknown): Uint8Array; readonly contentType?: string; } export interface KubeMQDeserializer { deserialize(data: Uint8Array, tags?: Record): 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: ```typescript 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 [#connection-lifecycle] The underlying `kubemq-js` client manages a long-lived gRPC connection whose state is reported as a `KubeMQStatus` enum: ```typescript title="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. ```typescript 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 [#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`. ## Related Topics [#related-topics]