Overview

A command handler is the receiving side of KubeMQ's Commands pattern — the code that actually does the work a caller is blocked waiting on. Instead of building your own request-routing layer on top of a queue, you register a handler once by collecting the Flow returned by subscribeToCommands, and KubeMQ delivers every matching command on that channel to it as a long-lived, server-streamed subscription, turning the channel into a synchronous RPC endpoint.

Handling happens inside the collect block: you read the command's body and tags, run your business logic (here, content-based — inspecting the body to decide the outcome), then build a reply with cmd.respond { executed = success; metadata = ...; body = ...; tags = ... } and send it with sendCommandResponse. Because respond is built from the received command, the correlation ID needed to route the reply back to the exact caller is carried automatically — you only set the fields that describe the outcome.

Gotchas: the reply must be sent within the caller's timeoutMs or the caller sees a timeout even if you eventually respond; tags set on the response are separate from the request's tags and propagate back to the sender, so don't assume they're echoed automatically; and a collector that never completes (no take(n) or cancellation) keeps the subscription — and the coroutine job — running indefinitely.

Prerequisites

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

Code

HandleCommandExample.kt

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

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

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

            // Handler that processes command body and builds response with metadata
            val handlerJob = launch {
                client.subscribeToCommands {
                    channel = CHANNEL
                }.take(2).collect { cmd ->
                    val body = String(cmd.body)
                    println("Handler received: $body (tags=${cmd.tags})")

                    // Process the command and build response
                    val success = body.contains("valid")
                    val response = cmd.respond {
                        executed = success
                        metadata = if (success) "processed-ok" else "rejected"
                        this.body = "Result for: $body".toByteArray()
                        tags = mapOf("handler" to "kotlin-handler-1")
                    }
                    client.sendCommandResponse(response)
                    println("Handler sent response: executed=$success")
                }
            }

            delay(500)

            // Send valid command
            val resp1 = client.sendCommand(commandMessage {
                channel = CHANNEL
                body = "valid-command".toByteArray()
                metadata = "test"
                timeoutMs = 10000
                tags["request-type"] = "test"
            })
            println("Command 1 result: executed=${resp1.executed}")

            // Send another command
            val resp2 = client.sendCommand(commandMessage {
                channel = CHANNEL
                body = "invalid-command".toByteArray()
                timeoutMs = 10000
            })
            println("Command 2 result: executed=${resp2.executed}")

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

How It Works

The handler uses cmd.respond { } DSL to build a rich response with executed, metadata, body, and tags.
Command processing can be content-based -- the handler inspects the command body to decide the response.
Tags are propagated bidirectionally between sender and handler.
The take(2) operator limits the handler to process exactly 2 commands before completing.

Handle Command

Overview

Prerequisites

Code

How It Works

On this page

Handle Command

Overview

Prerequisites

Code

How It Works

Related

On this page