Overview

A command is KubeMQ's fire-and-confirm RPC pattern: you reach for it when you need to know that an action actually ran on the other end — "restart the worker service" — but you don't need any data back, just a yes/no on execution. It's the middle ground between one-way pub/sub, where you get no confirmation at all, and a query, where the handler returns a result payload. Commands turn "I hope that worked" into a definite outcome your caller can branch on.

This sample builds that lesson: subscribeToCommands() registers a long-lived stream that invokes onReceiveCommandCallback on a background thread, and the handler acknowledges via client.sendResponseMessage(CommandResponseMessage.builder().commandReceived(cmd).isExecuted(true).build()). That call is what unblocks client.sendCommand(), which blocks the calling thread and returns a CommandResponseMessage whose isExecuted() reflects the outcome.

Gotchas: if no handler is subscribed (or it's still starting up), sendCommand() blocks for the full timeoutInSeconds before throwing KubeMQTimeoutException — there's no fast "nobody's listening" error, so set the timeout large enough for the handler's real processing time. A handler that never calls sendResponseMessage leaves the caller hanging until timeout. And a command's response carries no business data — if you need the handler to return a value, use a query instead.

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

SendCommandExample.java

package io.kubemq.example.commands;

import io.kubemq.sdk.cq.*;
import io.kubemq.sdk.common.ServerInfo;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class SendCommandExample {
    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-commands-send-command-client";
    private static final String CHANNEL = "java-commands.send-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();
        ServerInfo info = client.ping();
        System.out.println("Connected to: " + info.getHost());
        // Create the commands channel
        client.createCommandsChannel(CHANNEL);

        CountDownLatch latch = new CountDownLatch(1);

        // Subscribe to handle incoming commands (handler sends response)
        CommandsSubscription sub = CommandsSubscription.builder()
                .channel(CHANNEL)
                .onReceiveCommandCallback(cmd -> {
                    System.out.println("  Handler received: " + new String(cmd.getBody()));
                    client.sendResponseMessage(CommandResponseMessage.builder()
                            .commandReceived(cmd).isExecuted(true).build());
                    latch.countDown();
                })
                .onErrorCallback(err -> System.err.println("Error: " + err))
                .build();

        // Start the command handler subscription
        client.subscribeToCommands(sub);
        Thread.sleep(300);

        Map<String, String> tags = new HashMap<>();
        tags.put("action", "restart-service");

        CommandMessage command = CommandMessage.builder()
                .channel(CHANNEL).body("Restart the worker service".getBytes())
                .metadata("Command metadata").tags(tags).timeoutInSeconds(10).build();

        // Send a command and wait for the response
        CommandResponseMessage response = client.sendCommand(command);
        System.out.println("Command executed: " + response.isExecuted());

        latch.await(5, TimeUnit.SECONDS);
        // Clean up resources
        sub.cancel();
        client.deleteCommandsChannel(CHANNEL);
        client.close();
    }
}

// Expected output:
// Connected to: <host>
//   Handler received: Restart the worker service
// Command executed: true

How It Works

subscribeToCommands() registers a long-lived gRPC stream; when a command arrives the SDK calls onReceiveCommandCallback on a background thread.
The handler sends an acknowledgement via client.sendResponseMessage(CommandResponseMessage.builder().commandReceived(cmd).isExecuted(true).build()) — this is what unblocks the sendCommand() caller.
sendCommand() blocks the calling thread and returns CommandResponseMessage; isExecuted() is true when the handler sets it, or a KubeMQTimeoutException is thrown on timeout.
timeoutInSeconds in CommandMessage is required and must be large enough for the handler's expected processing time.

Send Command

Overview

Prerequisites

Code

How It Works

On this page

Send Command

Overview

Prerequisites

Code

How It Works

Related

On this page