Overview

A query handler is the answering side of KubeMQ's request/response RPC pattern — the code that does real work and sends back data, unlike a Command handler, which only acknowledges receipt. Reach for it whenever a caller needs an actual answer — a lookup result, a computed value, a status object — not just confirmation that a message arrived.

Registering a handler by collecting subscribeToQueries opens a subscription; the broker delivers every matching query to your collect block as it arrives. The handler builds a response with query.respond { ... }, which carries the original query's correlation id back to the broker so the answer routes to the specific caller blocked waiting, and sets body (plus optional metadata and tags) with the real result before calling sendQueryResponse.

Gotchas: if the handler never sends a response, the caller blocks until its own timeoutMs elapses and fails with a timeout, not a fast error. An exception inside the collect block doesn't automatically become a failure reply, so uncaught errors can leave the sender hanging. And because every matching query lands on the same collector, slow handler code delays every other in-flight caller.

Prerequisites

KubeMQ server running on localhost:50000
Kotlin SDK installed (implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))

Code

HandleQueryExample.kt

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-handle-query"
private const val CHANNEL = "kotlin-queries.handle-query"

fun main() = runBlocking {
    val client = KubeMQClient.cq {
        address = ADDRESS
        clientId = CLIENT_ID
    }

    client.use {
        try {
            client.createQueriesChannel(CHANNEL)

            // Handler that parses query body and builds rich response
            val handlerJob = launch {
                client.subscribeToQueries {
                    channel = CHANNEL
                }.take(1).collect { query ->
                    val requestBody = String(query.body)
                    println("Handler received query: $requestBody")
                    println("  Tags: ${query.tags}")

                    // Parse and process
                    val responseBody = """{"userId": 123, "name": "John Doe"}"""
                    val response = query.respond {
                        executed = true
                        metadata = "application/json"
                        body = responseBody.toByteArray()
                        tags = mapOf("status" to "200", "handler" to "kotlin-handler")
                    }
                    client.sendQueryResponse(response)
                    println("Handler sent response with tags: ${response.tags}")
                }
            }

            delay(500)

            // Send query
            val resp = client.sendQuery(queryMessage {
                channel = CHANNEL
                body = "getUser:123".toByteArray()
                metadata = "json"
                timeoutMs = 10000
            })
            println("Response body: ${String(resp.body)}")
            println("Response metadata: ${resp.metadata}")
            println("Response tags: ${resp.tags}")

            handlerJob.join()
        } finally {
            try { client.deleteQueriesChannel(CHANNEL) } catch (_: Exception) {}
        }
        println("Done.")
    }
}

How It Works

The handler parses the query body and builds a JSON response.
Response tags can carry metadata like HTTP status codes or handler identifiers.
The metadata field can indicate content type (e.g., application/json).
Both request and response support arbitrary tags for structured metadata.

Handle Query

Overview

Prerequisites

Code

How It Works

On this page

Handle Query

Overview

Prerequisites

Code

How It Works

Related

On this page