Request-Reply
Implement the request-reply pattern over KubeMQ queries, awaiting a synchronous response, using the Kotlin SDK.
Overview
Request-reply gives you synchronous RPC on top of KubeMQ's messaging fabric: a caller sends a query and blocks until the handler actually processing the request sends back a real answer — not just an acknowledgment. Reach for it whenever the caller needs a return value to proceed — a lookup, a computed result, a status check — the same shape as an HTTP call, but routed by KubeMQ instead of a service mesh or DNS.
A handler collects from subscribeToQueries and, for each request, builds a reply with request.respond { ... } before calling client.sendQueryResponse — this copies the request's correlation data so KubeMQ can route the response to the one caller waiting, not broadcast it. The caller's client.sendQuery suspends until that response lands or timeoutMs elapses, then returns a response carrying body and tags.
Gotchas: if no subscriber is listening — or the handler crashes before replying — sendQuery simply times out; there's no way to distinguish "no handler" from "handler is slow" from the timeout alone. Always build the reply from request.respond { ... } rather than a fresh object — that's what carries the correlation data back, and skipping it silently drops or misroutes the response. If you don't actually need a return value, use commands instead — they only need an ack, so they don't tie up a caller waiting on a round trip.
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.patterns
import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.cq.queryMessage
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-patterns-request-reply"
private const val CHANNEL = "kotlin-patterns.request-reply"
fun main() = runBlocking {
val client = KubeMQClient.cq {
address = ADDRESS
clientId = CLIENT_ID
}
client.use {
try {
client.createQueriesChannel(CHANNEL)
// Service handler (receives request, sends response)
val serviceJob = launch {
client.subscribeToQueries {
channel = CHANNEL
}.collect { request ->
val requestBody = String(request.body)
println(" [Service] Received: $requestBody")
val responseBody = """{"userId": 123, "name": "John Doe"}"""
val response = request.respond {
executed = true
body = responseBody.toByteArray()
tags = mapOf("status" to "200")
}
client.sendQueryResponse(response)
}
}
delay(300)
// Client sends request and waits for reply
println("Client sending request...")
val response = client.sendQuery(queryMessage {
channel = CHANNEL
body = "getUser:123".toByteArray()
timeoutMs = 10000
})
println("Response: ${String(response.body)}")
println("Status: ${response.tags["status"]}")
serviceJob.cancel()
} finally {
try { client.deleteQueriesChannel(CHANNEL) } catch (_: Exception) {}
}
println("Done.")
}
}How It Works
- The service handler subscribes to a query channel and responds to each request.
- The client sends a query and suspends until the service responds or the timeout elapses.
- Response tags carry metadata (like HTTP status codes) alongside the response body.
- This pattern is equivalent to synchronous RPC over messaging infrastructure.
Related
Was this page helpful?