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 via CommandsSubscription.builder().onReceiveCommandCallback(...) and subscribeToCommands(), and KubeMQ delivers every matching command on that channel to it as a long-lived, server-streamed gRPC call, turning the channel into a synchronous RPC endpoint.

Handling happens inside the callback: you read the command's body, run your business logic, then build a reply with CommandResponseMessage.builder().commandReceived(cmd).isExecuted(success).build() and send it with client.sendResponseMessage(). Passing commandReceived(cmd) is what copies the correlation ID so the broker can route the reply back to the exact caller blocked on sendCommand() — nothing else identifies which request the response belongs to.

Gotchas: the reply must be sent within the caller's timeoutInSeconds or the caller sees a timeout even if you eventually respond; the SDK delivers commands sequentially on a background thread, so slow or blocking business logic head-of-line blocks the next command — dispatch to a thread pool if processing is non-trivial; and omitting sendResponseMessage() entirely leaves the sender waiting until it times out.

Prerequisites

KubeMQ server running on localhost:50000
Java SDK installed (implementation 'io.kubemq.sdk:kubemq-sdk-Java:3.1.1' (Gradle) or Maven dependency from Getting Started)

Code

HandleCommandExample.java

package io.kubemq.example.commands;

import io.kubemq.sdk.cq.*;
import java.util.concurrent.CountDownLatch;

public class HandleCommandExample {
    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-commands-handle-command-client";
    private static final String CHANNEL = "java-commands.handle-command";

    public static void main(String[] args) throws InterruptedException {
        // Create a client connected to the KubeMQ server
        CQClient client = CQClient.builder().address(ADDRESS).clientId(CLIENT_ID).build();
        client.ping();
        // Create the commands channel
        client.createCommandsChannel(CHANNEL);

        // Subscribe to handle incoming commands
        CommandsSubscription sub = CommandsSubscription.builder()
                .channel(CHANNEL)
                .onReceiveCommandCallback(cmd -> {
                    System.out.println("Handling command: " + new String(cmd.getBody()));
                    boolean success = processCommand(cmd);

                    CommandResponseMessage response = CommandResponseMessage.builder()
                            .commandReceived(cmd).isExecuted(success)
                            .error(success ? "" : "Processing failed").build();
                    // Send response back to the command sender
                    client.sendResponseMessage(response);
                    System.out.println("Response sent: executed=" + success);
                })
                .onErrorCallback(err -> System.err.println("Error: " + err))
                .build();

        // Start the command handler subscription
        client.subscribeToCommands(sub);
        System.out.println("Command handler listening on: " + CHANNEL);
        Thread.sleep(300);

        // Send a command and wait for the response
        CommandResponseMessage resp = client.sendCommand(CommandMessage.builder()
                .channel(CHANNEL).body("Process order #123".getBytes()).timeoutInSeconds(10).build());
        System.out.println("Result: " + resp.isExecuted());

        // Clean up resources
        sub.cancel();
        client.deleteCommandsChannel(CHANNEL);
        client.close();
    }

    private static boolean processCommand(CommandMessageReceived cmd) {
        try { Thread.sleep(50); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        return true;
    }
}

How It Works

subscribeToCommands() opens a server-streaming gRPC call; the SDK delivers each inbound command to onReceiveCommandCallback sequentially on a background thread.
The handler must call client.sendResponseMessage() to acknowledge the command; omitting the response causes the sender's sendCommand() to time out.
CommandResponseMessage.builder().commandReceived(cmd) copies the correlation ID from the received command so KubeMQ routes the response back to the original caller.
processCommand() here simulates 50 ms of work; in production this is where you apply business logic — keep it non-blocking or dispatch to a thread pool to avoid head-of-line blocking on the subscription stream.

Handle Command

Overview

Prerequisites

Code

How It Works

On this page

Handle Command

Overview

Prerequisites

Code

How It Works

Related

On this page