Cached Query
Use server-side query response caching with cacheKey and cacheTtlSeconds for KubeMQ queries in Kotlin.
Overview
Query response caching lets the broker answer repeat requests without re-running your handler — useful when a query is expensive to compute (a database lookup, an aggregation, a downstream call) but the same input is asked for repeatedly in a short window. Only the first request pays the processing cost; every other caller gets the same answer straight from the broker.
Set cacheKey and cacheTtlSeconds on the query. The first query with a given key is a miss: it reaches the handler, and the broker stores the response under that key for the TTL. A subsequent query with the same key is a hit — the broker returns the stored response directly without invoking the handler. cacheHit on the response tells you which happened.
Gotchas: the cache is keyed by the string you choose, not by the query body — if the underlying data changes mid-TTL, callers can get a stale answer until it expires. Keys are scoped per channel, so the same key on another channel is a separate entry. Caching only helps when requests genuinely repeat with the same key.
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-cached-query"
private const val CHANNEL = "kotlin-queries.cached-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(2).collect { query ->
println("Handler received query: ${String(query.body)}")
val response = query.respond {
executed = true
body = """{"user":"alice","age":30}""".toByteArray()
metadata = "from-handler"
}
client.sendQueryResponse(response)
}
}
delay(500)
// First query -- hits handler, result cached with key
val resp1 = client.sendQuery(queryMessage {
channel = CHANNEL
body = "get-user:alice".toByteArray()
timeoutMs = 10000
cacheKey = "user:alice"
cacheTtlSeconds = 60
})
println("Query 1: cacheHit=${resp1.cacheHit}, body=${String(resp1.body)}")
// Second query with same cache key -- may return cached result
val resp2 = client.sendQuery(queryMessage {
channel = CHANNEL
body = "get-user:alice".toByteArray()
timeoutMs = 10000
cacheKey = "user:alice"
cacheTtlSeconds = 60
})
println("Query 2: cacheHit=${resp2.cacheHit}, body=${String(resp2.body)}")
subJob.join()
} finally {
try { client.deleteQueriesChannel(CHANNEL) } catch (_: Exception) {}
}
println("Done.")
}
}How It Works
cacheKey = "user:alice"sets the server-side cache key for the query response.cacheTtlSeconds = 60sets the cache TTL to 60 seconds.- The first query hits the handler; the response is cached server-side.
- The second query with the same cache key may return the cached response (
cacheHit = true). - Caching reduces handler load for frequently repeated queries.
Related
Was this page helpful?