# Usage (/integrations/nestjs/how-to/usage)



## Overview [#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](/learn/rpc) (Commands and Queries),
[Events](/learn/events), [Events Store](/learn/events-store), and [Queues](/learn/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())`      |

<Callout type="info">
  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`):

  <RunKubeMQ ports="[50000, 9090]" />
</Callout>

## Commands (Request-Reply) [#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.

```typescript title="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`).

```typescript title="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:

```typescript
@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](https://github.com/kubemq-io/kubemq-nestjs).

## Queries (Request-Reply) [#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()`.

```typescript title="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.

```typescript title="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.

```typescript
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](https://github.com/kubemq-io/kubemq-nestjs).

## Events (Fire-and-Forget) [#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.

```typescript title="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`.

```typescript title="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.

```typescript
// 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](https://github.com/kubemq-io/kubemq-nestjs).

## Events Store (Persistent Events) [#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:

```typescript title="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`.

```typescript title="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`:

```typescript title="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](https://github.com/kubemq-io/kubemq-nestjs).

## Queues (Reliable Delivery) [#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:

```typescript title="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).

```typescript title="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.

```typescript title="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();
  }
}
```

<Callout type="warn">
  `ack()`, `nack()`, and `reQueue()` are only available when the handler is declared with `{ manualAck: true }`. Calling them in auto-ack mode throws.
</Callout>

See the `queues/send-receive`, `queues/ack-reject`, `queues/dead-letter-queue`, and `decorators/manual-ack` examples in the [integration repository](https://github.com/kubemq-io/kubemq-nestjs).

## Queue Message Metadata [#queue-message-metadata]

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

```typescript
// 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`:

```typescript title="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](https://github.com/kubemq-io/kubemq-nestjs).

## Decorator Options Summary [#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](/integrations/nestjs/reference/api#decorator-options).

```typescript
@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 })
```

<Callout type="info">
  A single provider class can hold handlers for several — even all five — patterns at once. The `decorators/all-handlers` example in the [integration repository](https://github.com/kubemq-io/kubemq-nestjs) registers a `@CommandHandler`, `@QueryHandler`, `@EventHandler`, `@EventStoreHandler`, and `@QueueHandler` in one service, with a sender that exercises each.
</Callout>

## Related [#related]

<Cards>
  <Card title="Module Configuration" href="/integrations/nestjs/how-to/module-configuration" description="forRoot / forRootAsync, register / registerAsync, forTest, and multi-broker setups." />

  <Card title="CQRS Bridge" href="/integrations/nestjs/how-to/cqrs-bridge" description="Route @nestjs/cqrs CommandBus, QueryBus, and EventBus over KubeMQ channels." />

  <Card title="API reference" href="/integrations/nestjs/reference/api" description="Decorators, contexts, decorator options, and the full export surface." />
</Cards>