Overview

This page documents the transport's API surface for sending and handling messages — the five handler decorators and the KubeMQRecord builder. It does not re-explain the patterns themselves: for what each pattern is, see RPC (Commands and Queries), Events, Events Store, and Queues.

All five patterns flow through the same two transport primitives: the KubeMQServer strategy you attach with connectMicroservice (inbound handlers) and the KubeMQClientProxy you inject to send messages (outbound). You never instantiate a different client per pattern. Instead you choose the pattern in two places:

On the handler — pick the decorator: @CommandHandler, @QueryHandler, @EventHandler, @EventStoreHandler, or @QueueHandler.
On the client — pick the verb and, when needed, wrap the payload in a KubeMQRecord. client.send() is request-reply (Command by default); client.emit() is fire-and-forget (Event by default). KubeMQRecord overrides the default type with .asQuery(), .asEventStore(), or .asQueue().

The table below summarizes how the client side selects each pattern. The rest of this page walks through all five.

Pattern	Handler decorator	Client call
Command	`@CommandHandler`	`client.send(channel, data)`
Query	`@QueryHandler`	`client.send(channel, new KubeMQRecord(data).asQuery())`
Event	`@EventHandler`	`client.emit(channel, data)`
Events Store	`@EventStoreHandler`	`client.emit(channel, new KubeMQRecord(data).asEventStore())`
Queue	`@QueueHandler`	`client.emit(channel, new KubeMQRecord(data).asQueue())`

The transport speaks gRPC to the broker on port 50000 via the native kubemq-js SDK — no HTTP connector is involved. If you do not have a broker yet, start one locally with Docker (gRPC on 50000, dashboard and REST 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

Commands (Request-Reply)

Commands are true request-reply with a server-enforced timeout: the client blocks until the handler returns a value (the response) or the timeout elapses. The default command timeout is 10 seconds; override it per-server with defaultCommandTimeout on KubeMQServer.

On the client, send returns an RxJS Observable, so wrap it in firstValueFrom to await the single response.

command.service.ts

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

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

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

  async createOrder(): Promise<void> {
    await this.client.connect();
    const response = await firstValueFrom(
      this.client.send('orders.create', { action: 'create-user', name: 'Alice' }),
    );
    this.logger.log(`Command response: ${JSON.stringify(response)}`);
  }
}

The handler binds the channel with @CommandHandler. Returning a value from the method sends it back as the command response. The injected KubeMQCommandContext exposes request-reply metadata including fromClientId (the sender's client ID) and channel, alongside the base context fields (id, timestamp, tags, patternType).

command.handler.ts

import { Injectable, Logger } from '@nestjs/common';
import { CommandHandler, KubeMQCommandContext } from '@kubemq/nestjs-transport';

@Injectable()
export class CommandHandlerService {
  private readonly logger = new Logger('CommandHandler');

  @CommandHandler('orders.create')
  async handle(data: Record<string, unknown>, ctx: KubeMQCommandContext) {
    this.logger.log(`Command from ${ctx.fromClientId} on ${ctx.channel}`);
    return { executed: true, action: data.action };
  }
}

Pass a group to load-balance commands across a consumer group, so multiple handler replicas share delivery instead of each receiving a copy:

@CommandHandler('orders.create', { group: 'order-writers' })

See the runnable examples rpc/send-command, rpc/handle-command, rpc/command-timeout, and rpc/command-group in the integration repository.

Queries (Request-Reply)

Queries are request-reply like Commands, but semantically a read: the responder returns data without mutating state. Because both Commands and Queries use client.send(), a Query must be marked explicitly by wrapping the payload in a KubeMQRecord and calling .asQuery().

query.service.ts

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

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

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

  async getOrder(id: string): Promise<void> {
    await this.client.connect();
    const order = await firstValueFrom(
      this.client.send('orders.get', new KubeMQRecord({ id }).asQuery()),
    );
    this.logger.log(`Query response: ${JSON.stringify(order)}`);
  }
}

The handler uses @QueryHandler and receives a KubeMQQueryContext (the same request-reply context as Commands, with fromClientId and replyChannel). Like Commands, the returned value is the response.

query.handler.ts

import { Injectable, Logger } from '@nestjs/common';
import { QueryHandler, KubeMQQueryContext } from '@kubemq/nestjs-transport';

@Injectable()
export class QueryHandlerService {
  private readonly logger = new Logger('QueryHandler');

  @QueryHandler('orders.get')
  async handle(data: { id: string }, ctx: KubeMQQueryContext) {
    this.logger.log(`Processing query on ${ctx.channel}: ${JSON.stringify(data)}`);
    return { orderId: data.id, name: 'Widget', total: 29.99 };
  }
}

KubeMQ supports server-side query caching: attach a cacheKey and cacheTtl (seconds) via withMetadata, and the broker returns the cached response for repeat queries within the TTL window instead of re-invoking the handler.

const record = new KubeMQRecord({ configKey: 'app.settings' })
  .asQuery()
  .withMetadata({ cacheKey: 'config:app.settings', cacheTtl: 60 });

See the rpc/send-query, rpc/handle-query, and rpc/cached-query examples in the integration repository.

Events (Fire-and-Forget)

Events are one-way: the publisher fires and does not wait for a response. client.emit() sends an Event by default, so no KubeMQRecord is required.

event.service.ts

await firstValueFrom(
  this.client.emit('orders.updated', { id: 'order-123', status: 'shipped' }),
);

The handler uses @EventHandler and receives the base KubeMQContext (Events carry no reply channel or sequence). The method returns void.

event.handler.ts

import { Injectable, Logger } from '@nestjs/common';
import { EventHandler, KubeMQContext } from '@kubemq/nestjs-transport';

@Injectable()
export class EventHandlerService {
  private readonly logger = new Logger('EventHandler');

  @EventHandler('orders.updated')
  async handleOrderUpdated(
    data: { id: string; status: string },
    ctx: KubeMQContext,
  ): Promise<void> {
    this.logger.log(`Order ${data.id} updated to ${data.status} on ${ctx.channel}`);
  }
}

Events support several delivery shapes:

Multiple subscribers — every active subscriber on a channel receives a copy (fanout).
Consumer groups — pass a group so members of the group load-balance delivery instead of each getting a copy.
Array-of-channels subscription — a single handler can subscribe to several channels at once.

// Load-balanced across the "notifiers" group
@EventHandler('orders.updated', { group: 'notifiers' })

// One handler, multiple channels
@EventHandler(['orders.updated', 'orders.created'])

See the events/basic-pubsub, events/consumer-group, and events/multiple-subscribers examples in the integration repository.

Events Store (Persistent Events)

Events Store is a persistent, ordered, replayable event stream. Unlike fire-and-forget Events, messages are retained by the broker and assigned a monotonically increasing sequence number, so a subscriber can replay history from a chosen start position. Mark the message with .asEventStore() on the client:

event-store.service.ts

const record = new KubeMQRecord({
  orderId: 'order-123',
  action: 'shipped',
  timestamp: new Date().toISOString(),
}).asEventStore();

await firstValueFrom(this.client.emit('orders.history', record));

The handler declares where its subscription begins via the startFrom option, and reads the per-message sequence from KubeMQEventStoreContext.sequence.

event-store.handler.ts

import { Injectable, Logger } from '@nestjs/common';
import { EventStoreHandler, KubeMQEventStoreContext } from '@kubemq/nestjs-transport';

@Injectable()
export class EventStoreHandlerService {
  private readonly logger = new Logger('EventStoreHandler');

  @EventStoreHandler('orders.history', { startFrom: 'first' })
  async handleOrderHistory(data: unknown, ctx: KubeMQEventStoreContext): Promise<void> {
    this.logger.log(`Event #${ctx.sequence} on ${ctx.channel}: ${JSON.stringify(data)}`);
  }
}

To replay from a specific point, combine startFrom: 'sequence' (or 'time') with startValue:

event-store.handler.ts

@EventStoreHandler('orders.history', { startFrom: 'sequence', startValue: 1 })
async handleEvent(data: unknown, ctx: KubeMQEventStoreContext): Promise<void> {
  this.logger.log(`Replayed event (seq=${ctx.sequence}): ${JSON.stringify(data)}`);
}

The startFrom option (EventStoreHandlerOptions.startFrom) accepts these positions, either as a string or the equivalent numeric 1–6:

Start position	Numeric	Meaning
`'new'`	`1`	Only events arriving after subscription (default)
`'first'`	`2`	Replay from the first stored event
`'last'`	`3`	Start from the last stored event
`'sequence'`	`4`	Start from the `startValue` sequence number
`'time'`	`5`	Start from the `startValue` time
`'timeDelta'`	`6`	Start from now minus `startValue`

startValue supplies the sequence number, absolute time, or time delta required by 'sequence', 'time', and 'timeDelta'.

See the events-store/replay-from-sequence, events-store/start-from-first, events-store/start-from-last, and events-store/start-new-only examples in the integration repository.

Queues (Reliable Delivery)

Queues provide reliable, at-least-once delivery with visibility timeout, redelivery, and dead-letter routing. Mark the message with .asQueue() on the client:

queue.service.ts

const record = new KubeMQRecord({ orderId: 'order-123' }).asQueue();
await firstValueFrom(this.client.emit('orders.process', record));

By default a @QueueHandler runs in auto-ack mode: the message is acknowledged once the handler returns normally and is redelivered if it throws. The KubeMQQueueContext exposes sequence (the message's position in the queue) and receiveCount (how many times this message has been delivered).

queue.handler.ts

import { Injectable, Logger } from '@nestjs/common';
import { QueueHandler, KubeMQQueueContext } from '@kubemq/nestjs-transport';

@Injectable()
export class QueueHandlerService {
  private readonly logger = new Logger('QueueHandler');

  @QueueHandler('orders.process', { maxMessages: 1 })
  async handleProcessOrder(data: unknown, ctx: KubeMQQueueContext): Promise<void> {
    this.logger.log(`Queue message #${ctx.sequence}, delivery #${ctx.receiveCount}`);
  }
}

For explicit control, set manualAck: true and call ctx.ack(), ctx.nack(), or ctx.reQueue(channel) yourself. ack() confirms processing, nack() rejects (the message is redelivered or dead-lettered per the queue's policy), and reQueue(channel) re-routes the message to another channel — useful as an application-level dead-letter path.

queue.handler.ts

import { Injectable, Logger } from '@nestjs/common';
import { QueueHandler, KubeMQQueueContext } from '@kubemq/nestjs-transport';

@Injectable()
export class ProcessHandlerService {
  private readonly logger = new Logger('ProcessHandler');

  @QueueHandler('orders.process', { manualAck: true })
  async handleProcessOrder(data: { type: string }, ctx: KubeMQQueueContext): Promise<void> {
    if (data.type === 'valid') {
      ctx.ack();
      return;
    }
    if (data.type === 'retry') {
      ctx.reQueue('orders.dlq');
      return;
    }
    ctx.nack();
  }
}

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

See the queues/send-receive, queues/ack-reject, queues/dead-letter-queue, and decorators/manual-ack examples in the integration repository.

Queue Message Metadata

Queue messages carry a policy for time-to-live and delayed delivery. Chain .withMetadata() after .asQueue():

// TTL and delayed delivery
this.client.emit('orders.process', new KubeMQRecord(data).asQueue().withMetadata({
  expirationSeconds: 300,
  delaySeconds: 10,
}));

The runnable examples express the same options through the nested policy object, which also carries dead-letter routing via maxReceiveCount and maxReceiveQueue:

queue.service.ts

// Delayed delivery — visible after 5 seconds
const delayed = new KubeMQRecord({ task: 'scheduled-job' })
  .asQueue()
  .withMetadata({ policy: { delaySeconds: 5 } });

// Dead-letter after the first failed delivery
const poison = new KubeMQRecord({ action: 'poison' })
  .asQueue()
  .withMetadata({
    policy: {
      maxReceiveCount: 1,
      maxReceiveQueue: 'orders.process.dlq',
    },
  });

See the queues/delayed-messages and queues/batch-send examples in the integration repository.

Decorator Options Summary

Every decorator accepts the shared base options plus pattern-specific extras. The full option surface is defined in EventStoreHandlerOptions, QueueHandlerOptions, and the other interfaces.

Decorator	Shared options	Pattern-specific options
`@CommandHandler`	`group`, `maxConcurrent` / `concurrency`	—
`@QueryHandler`	`group`, `maxConcurrent` / `concurrency`	—
`@EventHandler`	`group`, `maxConcurrent` / `concurrency`	—
`@EventStoreHandler`	`group`, `maxConcurrent` / `concurrency`	`startFrom`, `startValue`
`@QueueHandler`	`group`, `maxConcurrent` / `concurrency`	`manualAck`, `maxMessages`, `waitTimeoutSeconds`, `batch`

group joins a consumer group for load-balanced delivery. maxConcurrent (aliased by concurrency, which takes precedence when both are set) caps concurrent handler executions. For the complete surface — including validation, dead-letter, and idempotency options — see the decorator API reference.

@CommandHandler('orders.create', { group: 'writers', maxConcurrent: 5 })

@EventStoreHandler('orders.history', { startFrom: 'sequence', startValue: 100 })

@QueueHandler('orders.process', { manualAck: true, maxMessages: 5, waitTimeoutSeconds: 60 })

@QueueHandler('orders.process', { batch: true, maxMessages: 10 })

A single provider class can hold handlers for several — even all five — patterns at once. The decorators/all-handlers example in the integration repository registers a @CommandHandler, @QueryHandler, @EventHandler, @EventStoreHandler, and @QueueHandler in one service, with a sender that exercises each.

Usage

Overview

Commands (Request-Reply)

Queries (Request-Reply)

Events (Fire-and-Forget)

Events Store (Persistent Events)

Queues (Reliable Delivery)

Queue Message Metadata

Decorator Options Summary

Module Configuration

CQRS Bridge

API reference

On this page