Send Query
Send a KubeMQ query to a handler and receive response data with metadata and tags using the Kotlin SDK.
Overview
This tutorial builds the RPC half of KubeMQ's request/reply patterns: a query, where the caller suspends for a handler's data payload instead of just a completion status. Reach for it whenever a caller needs an answer — fetching a record, running a lookup, or asking another service to compute a value on demand. You'll run a handler and a sender in the same process to see the full round trip.
client.sendQuery(queryMessage { ... }) sends a query and suspends until a response arrives. client.subscribeToQueries { } collects each incoming QueryReceived; the handler calls query.respond { } to build a reply with body, metadata, and tags, then client.sendQueryResponse(response) sends it back — KubeMQ correlates the reply to this call automatically, so the caller never tracks request IDs itself.
Gotchas: timeoutMs must cover however long the handler takes to run — a slow handler times out the caller even though the handler eventually succeeds. No handler subscribed yet also times out rather than erroring immediately, so startup order matters. resp.cacheHit only applies with server-side caching configured — for a plain query it's always false.
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.queries
import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.cq.queryMessage
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-queries-send-query"
private const val CHANNEL = "kotlin-queries.send-query"
fun main() = runBlocking {
val client = KubeMQClient.cq {
address = ADDRESS
clientId = CLIENT_ID
}
client.use {
try {
client.createQueriesChannel(CHANNEL)
// Subscribe to queries and respond
val subJob = launch {
client.subscribeToQueries {
channel = CHANNEL
}.take(1).collect { query ->
println("Received query: ${String(query.body)}")
val response = query.respond {
executed = true
metadata = "result"
body = """{"user":"alice","age":30}""".toByteArray()
}
client.sendQueryResponse(response)
println("Sent response for query ${query.id}")
}
}
delay(500)
// Send a query
val resp = client.sendQuery(queryMessage {
channel = CHANNEL
body = "get-user:alice".toByteArray()
metadata = "user-lookup"
timeoutMs = 10000
tags["request-type"] = "lookup"
})
println("Query result: ${String(resp.body)}, cacheHit=${resp.cacheHit}")
subJob.join()
} finally {
try { client.deleteQueriesChannel(CHANNEL) } catch (_: Exception) {}
}
println("Done.")
}
}How It Works
sendQuery(queryMessage { })sends a query and suspends until a response is received.- The handler receives a
QueryReceivedand builds a response withbody,metadata, andtags. resp.cacheHitindicates whether the response came from the server cache.- Queries differ from commands by returning data in the response body.
Related
Was this page helpful?