CommandMessage

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

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

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

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

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

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

CommandsSubscription

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

QueriesSubscription

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

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

Response Types

CommandResponseMessage

Sent from a command handler back to the sender.

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

QueryResponseMessage

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

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

`send_command(message)`

Sends a command and waits for confirmation.

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)`

Sends a query and waits for a data response.

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)`

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

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)`

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

Yields: QueryReceived with the same attributes as CommandReceived.

`send_response(response)`

Sends a response to a received command or query.

client.send_response(response_message)

Raises: ValidationError if request_id or reply_channel is missing.

RPC

On this page