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

Commands

sendCommand

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

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

suspend fun sendCommandResponse(message: CommandResponseMessage)

Sends the response back to the command sender.

commandMessage DSL

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

Command respond DSL

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

Queries

sendQuery

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

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

Same pattern as commands with query-specific response types.

queryMessage DSL

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

Quick Usage

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}")
}

RPC