# RPC (/sdks/kotlin/reference/rpc)



`CQClient` handles request-reply workloads. Commands embed timeouts in `commandMessage { }` DSL; queries extend with cache hints via `queryMessage { }`.

## Commands [#commands]

### sendCommand [#sendcommand]

```kotlin
suspend fun sendCommand(message: CommandMessage): CommandResponse
```

Sends a command and suspends until a worker responds or the timeout elapses.

**Parameters:**

| Name      | Type             | Required | Description                        |
| --------- | ---------------- | -------- | ---------------------------------- |
| `message` | `CommandMessage` | Yes      | Built via `commandMessage { }` DSL |

**Returns:** `CommandResponse`

| Field       | Type      | Description                                             |
| ----------- | --------- | ------------------------------------------------------- |
| `executed`  | `Boolean` | `true` if the handler executed the command successfully |
| `error`     | `String`  | Error message from the handler; empty on success        |
| `requestId` | `String`  | Matches the sent command's request ID                   |
| `timestamp` | `Instant` | Server timestamp of the response                        |

### subscribeToCommands [#subscribetocommands]

```kotlin
fun subscribeToCommands(config: CommandsSubscriptionConfig.() -> Unit): Flow<CommandReceived>
```

Returns a `Flow` of received commands. Each `CommandReceived` provides a `respond { }` DSL to build the response.

**Configuration DSL:**

| Property  | Type     | Required | Description          |
| --------- | -------- | -------- | -------------------- |
| `channel` | `String` | Yes      | Channel to listen on |
| `group`   | `String` | No       | Consumer group       |

### sendCommandResponse [#sendcommandresponse]

```kotlin
suspend fun sendCommandResponse(message: CommandResponseMessage)
```

Sends the response back to the command sender.

### commandMessage DSL [#commandmessage-dsl]

```kotlin
commandMessage {
    channel = "commands.shutdown"
    body = "graceful".toByteArray()
    metadata = "ops"
    timeoutMs = 10_000
    tags["priority"] = "high"
}
```

### Command respond DSL [#command-respond-dsl]

```kotlin
val response = cmd.respond {
    executed = true
    metadata = "processed"
    body = "result".toByteArray()
    tags = mapOf("handler" to "worker-1")
}
```

## Queries [#queries]

### sendQuery [#sendquery]

```kotlin
suspend fun sendQuery(message: QueryMessage): QueryResponse
```

Sends a query and suspends until a handler responds or the timeout elapses.

**QueryMessage** extends CommandMessage with cache hints:

| Property          | Type     | Description                  |
| ----------------- | -------- | ---------------------------- |
| `cacheKey`        | `String` | Cache key for response reuse |
| `cacheTtlSeconds` | `Int`    | Cache TTL in seconds         |

### subscribeToQueries [#subscribetoqueries]

```kotlin
fun subscribeToQueries(config: QueriesSubscriptionConfig.() -> Unit): Flow<QueryReceived>
```

Same pattern as commands with query-specific response types.

### queryMessage DSL [#querymessage-dsl]

```kotlin
queryMessage {
    channel = "queries.user-lookup"
    body = """{"id": 42}""".toByteArray()
    timeoutMs = 10_000
    cacheKey = "user:42"
    cacheTtlSeconds = 60
}
```

## Quick Usage [#quick-usage]

```kotlin title="Rpc.kt"
val cq = KubeMQClient.cq {
    address = "localhost:50000"
    clientId = "rpc-demo"
}

cq.use {
    val resp = cq.sendCommand(commandMessage {
        channel = "svc.commands"
        body = "restart".toByteArray()
        timeoutMs = 5000
    })
    println("Executed: ${resp.executed}")
}
```

## See Also [#see-also]

* [RPC Examples](/sdks/kotlin/how-to/rpc/)
* [RPC pattern](/learn/rpc/getting-started)
* [Kotlin SDK Getting Started](/sdks/kotlin)