Send Command
Send a KubeMQ command to a handler and receive the execution result using the Kotlin SDK RPC client.
Overview
A command is KubeMQ's fire-and-confirm RPC pattern: you reach for it when you need to know that an action actually ran on the other end — "restart-service" — but you don't need any data back, just a yes/no on execution. It's the middle ground between one-way pub/sub, where you get no confirmation at all, and a query, where the handler returns a result payload. Commands turn "I hope that worked" into a definite outcome your caller can branch on.
This sample builds that lesson: client.sendCommand(commandMessage { }) suspends until a response arrives or timeoutMs expires. The handler receives a CommandReceived, builds a reply with the respond { } DSL, and sends it back via client.sendCommandResponse(response) — that call is what resumes the sender's coroutine with the execution result.
Gotchas: if no handler is subscribed (or it's still starting up), sendCommand suspends for the full timeoutMs before failing — there's no fast "nobody's listening" error. A handler that never calls sendCommandResponse leaves the caller suspended until timeout. And a command's response carries no business data — if you need the handler to return a value, use a query instead.
Prerequisites
- KubeMQ server running on
localhost:50000 - Kotlin SDK installed (
implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))
Code
package io.kubemq.sdk.examples.commands
import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.cq.commandMessage
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-commands-send-command"
private const val CHANNEL = "kotlin-commands.send-command"
fun main() = runBlocking {
val client = KubeMQClient.cq {
address = ADDRESS
clientId = CLIENT_ID
}
client.use {
try {
client.createCommandsChannel(CHANNEL)
// Subscribe to commands and respond
val subJob = launch {
client.subscribeToCommands {
channel = CHANNEL
}.take(1).collect { cmd ->
println("Received command: ${String(cmd.body)}")
val response = cmd.respond {
executed = true
metadata = "processed"
}
client.sendCommandResponse(response)
println("Sent response for command ${cmd.id}")
}
}
delay(500)
// Send a command
val response = client.sendCommand(commandMessage {
channel = CHANNEL
body = "restart-service".toByteArray()
metadata = "ops"
timeoutMs = 10000
})
println("Command result: executed=${response.executed}, error=${response.error}")
subJob.join()
} finally {
try { client.deleteCommandsChannel(CHANNEL) } catch (_: Exception) {}
}
println("Done.")
}
}How It Works
sendCommand(commandMessage { })sends a command and suspends until a response is received or the timeout expires.- The handler receives a
CommandReceivedand builds a response using therespond { }DSL. sendCommandResponse(response)sends the response back to the caller.- The
timeoutMsproperty sets the maximum wait time for the response.
Related
Was this page helpful?