# Commands & Queries Reference (/learn/rpc/reference)



## Request Model [#request-model]

Commands and queries share the same request structure. The `RequestTypeData` field determines whether the request is treated as a command or query.

<TypeTable
  type="{
  RequestID: { type: 'string', description: 'Unique request identifier. Auto-generated (NUID) if empty.' },
  Channel: { type: 'string', description: 'Target channel where a responder is listening.', required: true },
  ClientID: { type: 'string', description: 'Identifier of the sending client. Must match ^[a-zA-Z0-9_-]+$.', required: true },
  RequestTypeData: { type: 'enum', description: 'Command or Query.', required: true },
  Metadata: { type: 'string', description: 'Text metadata. At least one of Body or Metadata required.' },
  Body: { type: 'bytes', description: 'Binary payload. At least one of Body or Metadata required.' },
  Tags: { type: 'map<string, string>', description: 'Key-value pairs for filtering and routing.' },
  Timeout: { type: 'integer', description: 'Maximum wait time in milliseconds. Must be greater than zero.', required: true },
  CacheKey: { type: 'string', description: '(Queries only) Cache key for server-side response caching.' },
  CacheTTL: { type: 'integer', description: '(Queries only) Cache TTL in milliseconds. Required if CacheKey is set.' },
}"
/>

## Response Model [#response-model]

<TypeTable
  type="{
  RequestID: { type: 'string', description: 'The original request ID this response corresponds to.' },
  ReplyChannel: { type: 'string', description: 'Internal reply channel (auto-set by server, always stripped from responses).' },
  Metadata: { type: 'string', description: 'Response metadata. Stripped for commands, preserved for queries.' },
  Body: { type: 'bytes', description: 'Response payload. Stripped for commands, preserved for queries.' },
  Tags: { type: 'map<string, string>', description: 'Key-value pairs from the responder.' },
  Executed: { type: 'boolean', description: 'Whether the command/query was successfully executed.' },
  Error: { type: 'string', description: 'Error description if execution failed. Empty on success.' },
  CacheHit: { type: 'boolean', description: '(Queries only) Whether the response was served from cache.' },
  Timestamp: { type: 'integer', description: 'Response generation timestamp (Unix nanoseconds).' },
}"
/>

## Command vs Query Response Differences [#command-vs-query-response-differences]

| Response Field | Command                   | Query                  |
| -------------- | ------------------------- | ---------------------- |
| `Body`         | Always `nil` (stripped)   | Preserved              |
| `Metadata`     | Always `""` (stripped)    | Preserved              |
| `CacheHit`     | Always `false` (stripped) | Preserved              |
| `ReplyChannel` | Always `""` (stripped)    | Always `""` (stripped) |
| `Executed`     | Preserved                 | Preserved              |
| `Error`        | Preserved                 | Preserved              |

## Validation Rules [#validation-rules]

### Channel Name [#channel-name]

| Rule            | Constraint                | Error Code |
| --------------- | ------------------------- | ---------- |
| Required        | Cannot be empty           | 102        |
| No trailing dot | Cannot end with `.`       | 119        |
| No whitespace   | Cannot contain spaces     | 108        |
| No wildcards    | Cannot contain `*` or `>` | 107        |

Valid channel name regex: `^[^\s*>]+[^.]$`

### Client ID [#client-id]

| Rule         | Constraint                    | Error Code |
| ------------ | ----------------------------- | ---------- |
| Required     | Cannot be empty               | 101        |
| Alphanumeric | Must match `^[a-zA-Z0-9_-]+$` | —          |

### Request Content [#request-content]

At least one of `Body` or `Metadata` must be provided. If both are empty, the request is rejected with error code 115.

### Timeout [#timeout]

| Rule     | Constraint                | Error Code |
| -------- | ------------------------- | ---------- |
| Required | Must be greater than zero | 109        |

### Cache (Queries Only) [#cache-queries-only]

| Rule              | Constraint                                   | Error Code |
| ----------------- | -------------------------------------------- | ---------- |
| CacheTTL required | If `CacheKey` is set, `CacheTTL` must be > 0 | 116        |

## Subscription Model [#subscription-model]

<TypeTable
  type="{
  ClientID: { type: 'string', description: 'Unique responder identifier.', required: true },
  Channel: { type: 'string', description: 'Channel to listen for requests. No wildcards.', required: true },
  Group: { type: 'string', description: 'Queue group name for load balancing. Empty = all responders get every request.' },
  SubscribeType: { type: 'enum', description: 'commands or queries.', required: true },
}"
/>

### Consumer Groups [#consumer-groups]

When multiple responders specify the same `group` value on the same channel:

* Each request is delivered to exactly **one** member of the group (round-robin)
* When `group` is empty, every responder receives every request (fan-out)
* Groups are independent per channel
* There is no limit on the number of group members

See [Load Balancing](/learn/rpc/how-to/load-balancing) for examples.

## Caching Configuration [#caching-configuration]

<Accordions>
  <Accordion title="How Query Caching Works">
    KubeMQ provides server-side response caching for queries. When a query includes a `CacheKey` and `CacheTTL`:

    1. KubeMQ checks the in-memory cache for an entry matching `CacheKey`
    2. **Cache hit:** Returns the cached response with `CacheHit: true` (responder is not called)
    3. **Cache miss:** Routes the query to a responder, caches the response, and returns it with `CacheHit: false`

    | Setting            | Value                                              |
    | ------------------ | -------------------------------------------------- |
    | Storage            | In-memory TTL cache                                |
    | Default expiration | Per-entry, specified by `CacheTTL` in milliseconds |
    | Cleanup interval   | Every 10 seconds                                   |
    | Persistence        | None — cache is cleared on server restart          |

    Commands do **not** support caching — the `CacheKey` and `CacheTTL` fields are ignored for command requests.

    See [Query Caching](/learn/rpc/tutorials/query-caching) for a step-by-step tutorial.
  </Accordion>
</Accordions>

## Transport Protocols [#transport-protocols]

### gRPC [#grpc]

| Method                                            | Type          | Description                      |
| ------------------------------------------------- | ------------- | -------------------------------- |
| `SendRequest(Request) → Response`                 | Unary         | Send command or query            |
| `SendResponse(Response) → Empty`                  | Unary         | Send response back to requester  |
| `SubscribeToRequests(Subscribe) → stream Request` | Server stream | Subscribe to commands or queries |

Default port: `50000`

### REST [#rest]

| Method | Path                  | Description                                 |
| ------ | --------------------- | ------------------------------------------- |
| `POST` | `/send/request`       | Send command or query                       |
| `POST` | `/send/response`      | Send RPC response                           |
| `GET`  | `/subscribe/requests` | WebSocket: subscribe to commands or queries |

Default port: `9090`

REST subscription query parameters:

| Parameter        | Description             | Example          |
| ---------------- | ----------------------- | ---------------- |
| `client_id`      | Client identifier       | `my-responder`   |
| `channel`        | Channel name            | `orders.process` |
| `group`          | Load balancing group    | `workers`        |
| `subscribe_type` | `commands` or `queries` | `commands`       |

## Internal Channel Mapping [#internal-channel-mapping]

| Pattern  | Channel Prefix | Example                      |
| -------- | -------------- | ---------------------------- |
| Commands | `_COMMANDS_.`  | `_COMMANDS_.orders.process`  |
| Queries  | `_QUERIES_.`   | `_QUERIES_.inventory.lookup` |

## Middleware Chain [#middleware-chain]

### Command Sender [#command-sender]

```text
Request → Logging → Monitor → Metrics → broker request
```

### Query Sender [#query-sender]

```text
Request → Logging → Monitor → Cache → Metrics → broker request
```

### Receiver (Commands and Queries) [#receiver-commands-and-queries]

```text
broker queue-subscribe → Logging → reqCh delivery
```

## Error Codes [#error-codes]

### Input Validation [#input-validation]

| Code | Error             | Description                                      |
| ---- | ----------------- | ------------------------------------------------ |
| 101  | Invalid ClientID  | ClientID is empty                                |
| 102  | Invalid Channel   | Channel is empty                                 |
| 107  | Invalid Channel   | Channel contains wildcards (`*` or `>`)          |
| 108  | Invalid Channel   | Channel contains whitespace                      |
| 109  | Invalid Timeout   | Timeout is zero or negative                      |
| 115  | Invalid Request   | Both body and metadata are empty                 |
| 116  | Invalid CacheTTL  | CacheKey is set but CacheTTL is zero or negative |
| 117  | Invalid RequestID | RequestID is empty (on response)                 |
| 119  | Invalid Channel   | Channel ends with `.`                            |

### Runtime Errors [#runtime-errors]

| Code | Error                   | Description                                      |
| ---- | ----------------------- | ------------------------------------------------ |
| 206  | Invalid Request Type    | Request type is not Command or Query             |
| 207  | Invalid Subscribe Type  | Subscribe type is not Commands or Queries        |
| 301  | Request Timeout         | No reply received before timeout expired         |
| 302  | Connection Unavailable  | broker connection is down                        |
| 303  | Invalid Response Format | Reply data cannot be unmarshaled                 |
| 409  | Shutdown Mode           | Server is shutting down, all operations rejected |
| 412  | Access Denied           | Authorization denied for the resource            |

## Delivery Semantics [#delivery-semantics]

| Aspect             | Behavior                                                          |
| ------------------ | ----------------------------------------------------------------- |
| Delivery guarantee | **At-most-once** (request is sent once, no automatic retry)       |
| Ordering           | Requests are independent (no ordering guarantee between requests) |
| Timeout            | Sender blocks until response arrives or timeout expires           |
| Acknowledgment     | Implicit — response is the acknowledgment                         |

## SDK Quick Reference [#sdk-quick-reference]

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go
    // Send Command
    client.SendCommand(ctx, kubemq.NewCommand().
        SetChannel("ch").SetBody([]byte("data")).SetTimeout(10*time.Second))

    // Send Query
    client.SendQuery(ctx, kubemq.NewQuery().
        SetChannel("ch").SetBody([]byte("data")).SetTimeout(10*time.Second))

    // Send Query with Cache
    client.SendQuery(ctx, kubemq.NewQuery().
        SetChannel("ch").SetBody([]byte("data")).SetTimeout(10*time.Second).
        SetCacheKey("key").SetCacheTTL(60*time.Second))

    // Subscribe to Commands
    client.SubscribeToCommands(ctx, "ch", "group",
        kubemq.WithOnCommandReceive(handler),
        kubemq.WithOnError(errHandler))

    // Subscribe to Queries
    client.SubscribeToQueries(ctx, "ch", "group",
        kubemq.WithOnQueryReceive(handler),
        kubemq.WithOnError(errHandler))
    ```
  </Tab>

  <Tab value="Python">
    ```python
    # Send Command
    client.send_command(CommandMessage(
        channel="ch", body=b"data", timeout_in_seconds=10))

    # Send Query
    client.send_query(QueryMessage(
        channel="ch", body=b"data", timeout_in_seconds=10))

    # Send Query with Cache
    client.send_query(QueryMessage(
        channel="ch", body=b"data", timeout_in_seconds=10,
        cache_key="key", cache_ttl_in_seconds=60))

    # Subscribe to Commands
    client.subscribe_to_commands(CommandsSubscription(
        channel="ch", group="group",
        on_receive_command_callback=handler,
        on_error_callback=err_handler), cancel=cancel)

    # Subscribe to Queries
    client.subscribe_to_queries(QueriesSubscription(
        channel="ch", group="group",
        on_receive_query_callback=handler,
        on_error_callback=err_handler), cancel=cancel)
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript
    // Send Command
    await client.sendCommand({
      channel: "ch", body: Buffer.from("data"), timeoutInSeconds: 10 });

    // Send Query
    await client.sendQuery({
      channel: "ch", body: Buffer.from("data"), timeoutInSeconds: 10 });

    // Send Query with Cache
    await client.sendQuery({
      channel: "ch", body: Buffer.from("data"), timeoutInSeconds: 10,
      cacheKey: "key", cacheTTL: 60000 });

    // Subscribe to Commands
    client.subscribeToCommands({
      channel: "ch", group: "group",
      onCommand: handler, onError: errHandler });

    // Subscribe to Queries
    client.subscribeToQueries({
      channel: "ch", group: "group",
      onQuery: handler, onError: errHandler });
    ```
  </Tab>

  <Tab value="Java">
    ```java
    // Send Command
    client.sendCommandRequest(CommandMessage.builder()
        .channel("ch").body("data".getBytes()).timeout(10000).build());

    // Send Query
    client.sendQueryRequest(QueryMessage.builder()
        .channel("ch").body("data".getBytes()).timeout(10000).build());

    // Send Query with Cache
    client.sendQueryRequest(QueryMessage.builder()
        .channel("ch").body("data".getBytes()).timeout(10000)
        .cacheKey("key").cacheTTL(60000).build());

    // Subscribe to Commands
    client.subscribeToCommands(CommandsSubscription.builder()
        .channel("ch").group("group")
        .onReceiveCommandCallback(handler)
        .onErrorCallback(errHandler).build());

    // Subscribe to Queries
    client.subscribeToQueries(QueriesSubscription.builder()
        .channel("ch").group("group")
        .onReceiveQueryCallback(handler)
        .onErrorCallback(errHandler).build());
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    // Send Command
    await client.SendCommandAsync(new CommandMessage {
        Channel = "ch", Body = Encoding.UTF8.GetBytes("data"),
        Timeout = TimeSpan.FromSeconds(10) });

    // Send Query
    await client.SendQueryAsync(new QueryMessage {
        Channel = "ch", Body = Encoding.UTF8.GetBytes("data"),
        Timeout = TimeSpan.FromSeconds(10) });

    // Send Query with Cache
    await client.SendQueryAsync(new QueryMessage {
        Channel = "ch", Body = Encoding.UTF8.GetBytes("data"),
        Timeout = TimeSpan.FromSeconds(10),
        CacheKey = "key", CacheTTL = TimeSpan.FromSeconds(60) });

    // Subscribe to Commands
    await foreach (var cmd in client.SubscribeToCommandsAsync(
        new CommandsSubscription { Channel = "ch", Group = "group" })) { }

    // Subscribe to Queries
    await foreach (var q in client.SubscribeToQueriesAsync(
        new QueriesSubscription { Channel = "ch", Group = "group" })) { }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin
    // Send Command
    client.sendCommand(CommandMessage(
        channel = "ch", body = "data".toByteArray(), timeout = 10000))

    // Send Query
    client.sendQuery(QueryMessage(
        channel = "ch", body = "data".toByteArray(), timeout = 10000))

    // Send Query with Cache
    client.sendQuery(QueryMessage(
        channel = "ch", body = "data".toByteArray(), timeout = 10000,
        cacheKey = "key", cacheTTL = 60000))

    // Subscribe to Commands
    client.subscribeToCommands(
        channel = "ch", group = "group",
        onCommand = handler, onError = errHandler)

    // Subscribe to Queries
    client.subscribeToQueries(
        channel = "ch", group = "group",
        onQuery = handler, onError = errHandler)
    ```
  </Tab>

  <Tab value="C++">
    ```cpp
    // Send Command
    kubemq::CommandMessage cmd;
    cmd.channel = "ch"; cmd.body = "data"; cmd.timeout = 10000;
    client.sendCommand(cmd);

    // Send Query
    kubemq::QueryMessage query;
    query.channel = "ch"; query.body = "data"; query.timeout = 10000;
    client.sendQuery(query);

    // Send Query with Cache
    query.cacheKey = "key"; query.cacheTTL = 60000;
    client.sendQuery(query);

    // Subscribe to Commands
    client.subscribeToCommands("ch", "group", handler, errHandler);

    // Subscribe to Queries
    client.subscribeToQueries("ch", "group", handler, errHandler);
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    // Send Command
    let command = CommandBuilder::new()
        .channel("ch").body(b"data".to_vec())
        .timeout(Duration::from_secs(10)).build();
    client.send_command(command).await?;

    // Send Query
    let query = QueryBuilder::new()
        .channel("ch").body(b"data".to_vec())
        .timeout(Duration::from_secs(10)).build();
    client.send_query(query).await?;

    // Send Query with Cache
    let query = QueryBuilder::new()
        .channel("ch").body(b"data".to_vec())
        .timeout(Duration::from_secs(10))
        .cache_key("key").cache_ttl(Duration::from_secs(60)).build();
    client.send_query(query).await?;

    // Subscribe to Commands
    client.subscribe_to_commands("ch", "group", handler, None).await?;

    // Subscribe to Queries
    client.subscribe_to_queries("ch", "group", handler, None).await?;
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # Send Command
    msg = KubeMQ::CQ::CommandMessage.new(
      channel: "ch", body: "data", timeout: 10)
    client.send_command(msg)

    # Send Query
    msg = KubeMQ::CQ::QueryMessage.new(
      channel: "ch", body: "data", timeout: 10)
    client.send_query(msg)

    # Send Query with Cache
    msg = KubeMQ::CQ::QueryMessage.new(
      channel: "ch", body: "data", timeout: 10,
      cache_key: "key", cache_ttl: 60)
    client.send_query(msg)

    # Subscribe to Commands
    sub = KubeMQ::CQ::CommandsSubscription.new(channel: "ch", group: "group")
    client.subscribe_to_commands(sub, cancellation_token: cancel,
      on_error: err_handler) { |cmd| handler.call(cmd) }

    # Subscribe to Queries
    sub = KubeMQ::CQ::QueriesSubscription.new(channel: "ch", group: "group")
    client.subscribe_to_queries(sub, cancellation_token: cancel,
      on_error: err_handler) { |query| handler.call(query) }
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir
    # Send Command
    command = KubeMQ.Command.new(
      channel: "ch", body: "data", timeout: 10_000)
    KubeMQ.Client.send_command(client, command)

    # Send Query
    query = KubeMQ.Query.new(
      channel: "ch", body: "data", timeout: 10_000)
    KubeMQ.Client.send_query(client, query)

    # Send Query with Cache
    query = KubeMQ.Query.new(
      channel: "ch", body: "data", timeout: 10_000,
      cache_key: "key", cache_ttl: 60_000)
    KubeMQ.Client.send_query(client, query)

    # Subscribe to Commands
    KubeMQ.Client.subscribe_to_commands(client, "ch",
      group: "group", on_command: handler, on_error: err_handler)

    # Subscribe to Queries
    KubeMQ.Client.subscribe_to_queries(client, "ch",
      group: "group", on_query: handler, on_error: err_handler)
    ```
  </Tab>
</Tabs>

## Related [#related]

* [Getting Started](/learn/rpc/getting-started) — send your first command and query
* [Configure Timeouts](/learn/rpc/how-to/timeout-configuration) — per-request timeouts and retries
* [Load Balancing](/learn/rpc/how-to/load-balancing) — distribute requests across responders
* [Events Reference](/learn/events/reference) — for the fire-and-forget pattern
* [Queues Reference](/learn/queues/reference) — for guaranteed delivery