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



## CommandMessage [#commandmessage]

Outbound command message for fire-and-confirm RPC. The broker forwards the command to a subscriber and returns a response indicating execution status.

```ruby
cmd = KubeMQ::CQ::CommandMessage.new(
  channel: "commands.user.create",
  timeout: 5000,
  metadata: "create-user",
  body: '{"name": "Alice"}',
  tags: { "source" => "api" }
)
```

### Attributes [#attributes]

| Attribute  | Type                     | Default             | Description                      |
| ---------- | ------------------------ | ------------------- | -------------------------------- |
| `id`       | `String`                 | Auto-generated UUID | Unique message identifier        |
| `channel`  | `String`                 | Required            | Target channel name              |
| `timeout`  | `Integer`                | Required            | Response timeout in milliseconds |
| `metadata` | `String`                 | `nil`               | Arbitrary metadata               |
| `body`     | `String`                 | `nil`               | Message payload                  |
| `tags`     | `Hash{String => String}` | `{}`                | Key-value tags                   |

## QueryMessage [#querymessage]

Outbound query message for request/reply with data. Supports server-side response caching.

```ruby
query = KubeMQ::CQ::QueryMessage.new(
  channel: "queries.user.get",
  timeout: 10_000,
  metadata: "get-user",
  body: '{"user_id": 42}',
  cache_key: "user:42",
  cache_ttl: 60
)
```

### Attributes [#attributes-1]

| Attribute   | Type                     | Default             | Description                      |
| ----------- | ------------------------ | ------------------- | -------------------------------- |
| `id`        | `String`                 | Auto-generated UUID | Unique message identifier        |
| `channel`   | `String`                 | Required            | Target channel name              |
| `timeout`   | `Integer`                | Required            | Response timeout in milliseconds |
| `metadata`  | `String`                 | `nil`               | Arbitrary metadata               |
| `body`      | `String`                 | `nil`               | Message payload                  |
| `tags`      | `Hash{String => String}` | `{}`                | Key-value tags                   |
| `cache_key` | `String`                 | `nil`               | Server-side cache key            |
| `cache_ttl` | `Integer`                | `nil`               | Cache TTL in seconds             |

## Subscription Types [#subscription-types]

### CommandsSubscription [#commandssubscription]

```ruby
sub = KubeMQ::CQ::CommandsSubscription.new(channel: "commands.orders", group: nil)
```

### QueriesSubscription [#queriessubscription]

```ruby
sub = KubeMQ::CQ::QueriesSubscription.new(channel: "queries.users", group: nil)
```

Both accept `channel` (required) and `group` (optional consumer group).

## Response Types [#response-types]

### CommandResponseMessage [#commandresponsemessage]

Sent from a command handler back to the sender.

```ruby
response = KubeMQ::CQ::CommandResponseMessage.new(
  request_id: cmd.id,
  reply_channel: cmd.reply_channel,
  executed: true,
  error: nil
)
client.send_response(response)
```

### QueryResponseMessage [#queryresponsemessage]

Sent from a query handler back to the sender with data.

```ruby
response = KubeMQ::CQ::QueryResponseMessage.new(
  request_id: query.id,
  reply_channel: query.reply_channel,
  executed: true,
  body: '{"name": "Alice"}',
  metadata: "result"
)
client.send_response(response)
```

## CQClient Methods [#cqclient-methods]

### `send_command(message)` [#send_commandmessage]

Sends a command and waits for confirmation.

```ruby
response = client.send_command(cmd)
puts "Executed: #{response.executed}"
```

**Returns:** `CommandResponse` with `executed`, `error`, and `timestamp` attributes.

**Raises:** `ValidationError`, `TimeoutError`, `ClientClosedError`, `ConnectionError`

### `send_query(message)` [#send_querymessage]

Sends a query and waits for a data response.

```ruby
response = client.send_query(query)
puts "Data: #{response.body}, cache_hit: #{response.cache_hit}"
```

**Returns:** `QueryResponse` with `executed`, `body`, `metadata`, `cache_hit`, and `error` attributes.

### `subscribe_to_commands(subscription, cancellation_token:, on_error:, &block)` [#subscribe_to_commandssubscription-cancellation_token-on_error-block]

Subscribes to incoming commands. Process commands in the block and call `send_response` to reply.

```ruby
token = KubeMQ::CancellationToken.new
sub = KubeMQ::CQ::CommandsSubscription.new(channel: "commands.orders")

client.subscribe_to_commands(sub, cancellation_token: token) do |cmd|
  client.send_response(
    KubeMQ::CQ::CommandResponseMessage.new(
      request_id: cmd.id,
      reply_channel: cmd.reply_channel,
      executed: true
    )
  )
end
```

**Yields:** `CommandReceived` with `id`, `channel`, `metadata`, `body`, `reply_channel`, `tags`, `timeout`, and `client_id`.

### `subscribe_to_queries(subscription, cancellation_token:, on_error:, &block)` [#subscribe_to_queriessubscription-cancellation_token-on_error-block]

Subscribes to incoming queries. Process queries and return data via `send_response`.

**Yields:** `QueryReceived` with the same attributes as `CommandReceived`.

### `send_response(response)` [#send_responseresponse]

Sends a response to a received command or query.

```ruby
client.send_response(response_message)
```

**Raises:** `ValidationError` if `request_id` or `reply_channel` is missing.