What You Will Build

An order service that sends "process order" commands to a handler and checks execution status. You will learn how command responses differ from query responses and how to handle timeouts.

A command round-trip: the sender waits for an execution acknowledgment only — no data comes back.

Steps

Create a Command Responder

The responder subscribes to the orders.process channel, processes incoming commands, and sends back an execution status.

command_responder.go

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/kubemq-io/kubemq-go/v2"
)

func main() {
    ctx := context.Background()
    client, err := kubemq.NewClient(ctx,
        kubemq.WithAddress("localhost", 50000),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    _, err = client.SubscribeToCommands(ctx, "orders.process", "",
        kubemq.WithOnCommandReceive(func(cmd *kubemq.CommandReceive) {
            fmt.Printf("Processing order: %s\n", cmd.Body)
            resp := kubemq.NewCommandReply().
                SetRequestId(cmd.Id).
                SetResponseTo(cmd.ResponseTo).
                SetExecutedAt(time.Now())
            _ = client.SendCommandResponse(ctx, resp)
        }),
        kubemq.WithOnError(func(err error) {
            log.Println("Error:", err)
        }),
    )
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println("Responder ready on 'orders.process'...")
    <-ctx.Done()
}

command_responder.py

import time
from kubemq.cq import Client as CQClient
from kubemq.cq import CommandsSubscription, CommandReceived, CommandResponse, CancellationToken

def on_command(request: CommandReceived) -> None:
    print(f"Processing order: {request.body.decode('utf-8')}")
    client.send_response_message(
        CommandResponse(command_received=request, is_executed=True)
    )

client = CQClient(address="localhost:50000")
cancel = CancellationToken()
client.subscribe_to_commands(
    subscription=CommandsSubscription(
        channel="orders.process",
        on_receive_command_callback=on_command,
        on_error_callback=lambda e: print(f"Error: {e}"),
    ),
    cancel=cancel,
)
print("Responder ready on 'orders.process'...")
time.sleep(3600)

command_responder.js

const { KubeMQClient } = require("kubemq-js");

const client = new KubeMQClient({ address: "localhost:50000" });

client.subscribeToCommands({
  channel: "orders.process",
  onCommand: (cmd) => {
    console.log("Processing order:", Buffer.from(cmd.body).toString());
    client.sendCommandResponse({ requestId: cmd.id, isExecuted: true });
  },
  onError: (err) => console.error("Error:", err.message),
});

console.log("Responder ready on 'orders.process'...");

CommandResponder.java

CQClient client = CQClient.builder()
    .address("localhost:50000")
    .clientId("order-responder")
    .build();

client.subscribeToCommands(CommandsSubscription.builder()
    .channel("orders.process")
    .onReceiveCommandCallback(cmd -> {
        System.out.println("Processing order: " + new String(cmd.getBody()));
        return CommandResponseMessage.builder()
            .requestId(cmd.getId())
            .isExecuted(true)
            .build();
    })
    .onErrorCallback(err -> System.err.println("Error: " + err.getMessage()))
    .build());

System.out.println("Responder ready on 'orders.process'...");
Thread.sleep(3600000);
client.close();

CommandResponder.cs

await using var client = new KubeMQClient(new KubeMQClientOptions());
await client.ConnectAsync();

Console.WriteLine("Responder ready on 'orders.process'...");
await foreach (var cmd in client.SubscribeToCommandsAsync(
    new CommandsSubscription { Channel = "orders.process" }))
{
    Console.WriteLine($"Processing order: {Encoding.UTF8.GetString(cmd.Body.Span)}");
    await client.SendCommandResponseAsync(new CommandResponse
    {
        RequestId = cmd.Id, IsExecuted = true
    });
}

CommandResponder.kt

val client = CQClient("localhost:50000")

client.subscribeToCommands(
    channel = "orders.process",
    onCommand = { cmd ->
        println("Processing order: ${String(cmd.body)}")
        client.sendCommandResponse(requestId = cmd.id, isExecuted = true)
    },
    onError = { err -> System.err.println("Error: ${err.message}") }
)

println("Responder ready on 'orders.process'...")
Thread.sleep(3600000)
client.close()

command_responder.cpp

#include <kubemq/client.h>
#include <iostream>
#include <thread>

auto client = kubemq::CQClient("localhost:50000");

client.subscribeToCommands("orders.process", "",
    [&client](const kubemq::CommandReceive& cmd) {
        std::cout << "Processing order: " << cmd.body << std::endl;
        client.sendCommandResponse(cmd.id, true);
    },
    [](const std::string& err) {
        std::cerr << "Error: " << err << std::endl;
    }
);

std::cout << "Responder ready on 'orders.process'..." << std::endl;
std::this_thread::sleep_for(std::chrono::hours(1));

command_responder.rs

use kubemq::prelude::*;
use kubemq::CommandReplyBuilder;

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .build()
        .await?;

    let rc = client.clone();
    let _sub = client
        .subscribe_to_commands("orders.process", "", move |cmd| {
            let c = rc.clone();
            Box::pin(async move {
                println!("Processing order: {}", String::from_utf8_lossy(&cmd.body));
                let reply = CommandReplyBuilder::new()
                    .request_id(&cmd.id)
                    .response_to(&cmd.response_to)
                    .build();
                let _ = c.send_command_response(reply).await;
            })
        }, None)
        .await?;

    println!("Responder ready on 'orders.process'...");
    tokio::signal::ctrl_c().await.ok();
    client.close().await?;
    Ok(())
}

command_responder.rb

require "kubemq"

client = KubeMQ::CQClient.new(address: "localhost:50000", client_id: "order-responder")
cancel = KubeMQ::CancellationToken.new

sub = KubeMQ::CQ::CommandsSubscription.new(channel: "orders.process")
client.subscribe_to_commands(sub, cancellation_token: cancel,
                             on_error: ->(e) { puts "Error: #{e.message}" }) do |cmd|
  puts "Processing order: #{cmd.body}"
  response = KubeMQ::CQ::CommandResponseMessage.new(
    request_id: cmd.id,
    reply_channel: cmd.reply_channel,
    executed: true
  )
  client.send_response(response)
end

puts "Responder ready on 'orders.process'..."
cancel.wait

command_responder.exs

{:ok, client} =
  KubeMQ.Client.start_link(address: "localhost:50000", client_id: "order-responder")

{:ok, _sub} =
  KubeMQ.Client.subscribe_to_commands(client, "orders.process",
    on_command: fn cmd ->
      IO.puts("Processing order: #{cmd.body}")
      KubeMQ.CommandReply.new(
        request_id: cmd.id,
        response_to: cmd.reply_channel,
        executed: true
      )
    end,
    on_error: fn err -> IO.puts("Error: #{err.message}") end
  )

IO.puts("Responder ready on 'orders.process'...")
Process.sleep(:infinity)

Send a Command

Send a command with body, metadata, and a 10-second timeout.

send_command.go

resp, err := client.SendCommand(ctx, kubemq.NewCommand().
    SetChannel("orders.process").
    SetBody([]byte(`{"action":"create","orderId":"ORD-5678"}`)).
    SetMetadata("order.create").
    SetTags(map[string]string{"priority": "high"}).
    SetTimeout(10 * time.Second))
if err != nil {
    log.Fatal(err)
}
log.Printf("Executed: %v, Error: %s", resp.Executed, resp.Error)

send_command.py

from kubemq.cq import Client as CQClient, CommandMessage

with CQClient(address="localhost:50000") as client:
    response = client.send_command(
        CommandMessage(
            channel="orders.process",
            body=b'{"action":"create","orderId":"ORD-5678"}',
            metadata="order.create",
            tags={"priority": "high"},
            timeout_in_seconds=10,
        )
    )
    print(f"Executed: {response.is_executed}, Error: {response.error}")

send_command.js

const { KubeMQClient } = require("kubemq-js");

const client = new KubeMQClient({ address: "localhost:50000" });

const response = await client.sendCommand({
  channel: "orders.process",
  body: Buffer.from(JSON.stringify({ action: "create", orderId: "ORD-5678" })),
  metadata: "order.create",
  tags: { priority: "high" },
  timeoutInSeconds: 10,
});
console.log("Executed:", response.isExecuted, "Error:", response.error);

SendCommand.java

CommandResponseMessage response = client.sendCommandRequest(
    CommandMessage.builder()
        .channel("orders.process")
        .body("{\"action\":\"create\",\"orderId\":\"ORD-5678\"}".getBytes())
        .metadata("order.create")
        .tags("priority=high")
        .timeout(10000)
        .build());
System.out.println("Executed: " + response.isExecuted()
    + ", Error: " + response.getError());

SendCommand.cs

var response = await client.SendCommandAsync(new CommandMessage
{
    Channel = "orders.process",
    Body = Encoding.UTF8.GetBytes("{\"action\":\"create\",\"orderId\":\"ORD-5678\"}"),
    Metadata = "order.create",
    Tags = new Dictionary<string, string> { ["priority"] = "high" },
    Timeout = TimeSpan.FromSeconds(10)
});
Console.WriteLine($"Executed: {response.IsExecuted}, Error: {response.Error}");

SendCommand.kt

val response = client.sendCommand(CommandMessage(
    channel = "orders.process",
    body = """{"action":"create","orderId":"ORD-5678"}""".toByteArray(),
    metadata = "order.create",
    tags = mapOf("priority" to "high"),
    timeout = 10000
))
println("Executed: ${response.isExecuted}, Error: ${response.error}")

send_command.cpp

kubemq::CommandMessage cmd;
cmd.channel = "orders.process";
cmd.body = R"({"action":"create","orderId":"ORD-5678"})";
cmd.metadata = "order.create";
cmd.tags["priority"] = "high";
cmd.timeout = 10000;

auto response = client.sendCommand(cmd);
std::cout << "Executed: " << response.isExecuted
          << ", Error: " << response.error << std::endl;

send_command.rs

use kubemq::CommandBuilder;
use std::collections::HashMap;
use std::time::Duration;

let command = CommandBuilder::new()
    .channel("orders.process")
    .body(br#"{"action":"create","orderId":"ORD-5678"}"#.to_vec())
    .metadata("order.create")
    .tags(HashMap::from([("priority".to_string(), "high".to_string())]))
    .timeout(Duration::from_secs(10))
    .build();

let response = client.send_command(command).await?;
println!("Executed: {}, Error: '{}'", response.executed, response.error);

send_command.rb

msg = KubeMQ::CQ::CommandMessage.new(
  channel: "orders.process",
  body: '{"action":"create","orderId":"ORD-5678"}',
  metadata: "order.create",
  tags: { "priority" => "high" },
  timeout: 10_000 # milliseconds
)

response = client.send_command(msg)
puts "Executed: #{response.executed}, Error: #{response.error}"

send_command.exs

command =
  KubeMQ.Command.new(
    channel: "orders.process",
    body: ~s({"action":"create","orderId":"ORD-5678"}),
    metadata: "order.create",
    tags: %{"priority" => "high"},
    timeout: 10_000
  )

case KubeMQ.Client.send_command(client, command) do
  {:ok, response} ->
    IO.puts("Executed: #{response.executed}, Error: #{response.error}")

  {:error, err} ->
    IO.puts("Command failed: #{err.message}")
end

Check Execution Status

The response contains only Executed (boolean) and Error (string). Body and metadata are always stripped from command responses.

check_status.go

if resp.Executed {
    log.Println("Command executed successfully")
} else {
    log.Printf("Command failed: %s", resp.Error)
}

// Body is always nil for commands
log.Printf("Response body: %v", resp.Body) // nil

check_status.py

if response.is_executed:
    print("Command executed successfully")
else:
    print(f"Command failed: {response.error}")

# Body is always empty for commands
print(f"Response body: {response.body}")  # b''

check_status.js

if (response.isExecuted) {
  console.log("Command executed successfully");
} else {
  console.log("Command failed:", response.error);
}

// Body is always empty for commands
console.log("Response body:", response.body); // undefined

CheckStatus.java

if (response.isExecuted()) {
    System.out.println("Command executed successfully");
} else {
    System.out.println("Command failed: " + response.getError());
}

// Body is always null for commands
System.out.println("Response body: " + response.getBody()); // null

CheckStatus.cs

if (response.IsExecuted)
    Console.WriteLine("Command executed successfully");
else
    Console.WriteLine($"Command failed: {response.Error}");

// Body is always empty for commands
Console.WriteLine($"Response body: {response.Body.Length}"); // 0

CheckStatus.kt

if (response.isExecuted) {
    println("Command executed successfully")
} else {
    println("Command failed: ${response.error}")
}

// Body is always empty for commands
println("Response body: ${response.body?.size}") // null or 0

check_status.cpp

if (response.isExecuted) {
    std::cout << "Command executed successfully" << std::endl;
} else {
    std::cout << "Command failed: " << response.error << std::endl;
}

// Body is always empty for commands
std::cout << "Response body: " << response.body << std::endl; // ""

check_status.rs

if response.executed {
    println!("Command executed successfully");
} else {
    println!("Command failed: {}", response.error);
}

// CommandResponse carries no body — only command_id, executed, executed_at, error, tags
println!("Command id: {}", response.command_id);

check_status.rb

if response.executed
  puts "Command executed successfully"
else
  puts "Command failed: #{response.error}"
end

# CommandResponse carries no body — only request_id, executed, error, timestamp, tags
puts "Request id: #{response.request_id}"

check_status.exs

if response.executed do
  IO.puts("Command executed successfully")
else
  IO.puts("Command failed: #{response.error}")
end

# CommandResponse carries no body — only command_id, executed, executed_at, error, tags
IO.puts("Command id: #{response.command_id}")

Handle Timeout

When no responder replies within the timeout, the sender receives a timeout error (code 301).

handle_timeout.go

resp, err := client.SendCommand(ctx, kubemq.NewCommand().
    SetChannel("orders.process").
    SetBody([]byte("test")).
    SetTimeout(2 * time.Second))
if err != nil {
    log.Printf("Command failed: %v", err)
    return
}
if !resp.Executed {
    log.Printf("Timeout or error: %s", resp.Error)
}

handle_timeout.py

try:
    response = client.send_command(
        CommandMessage(
            channel="orders.process",
            body=b"test",
            timeout_in_seconds=2,
        )
    )
    if not response.is_executed:
        print(f"Timeout or error: {response.error}")
except Exception as e:
    print(f"Command failed: {e}")

handle_timeout.js

try {
  const response = await client.sendCommand({
    channel: "orders.process",
    body: Buffer.from("test"),
    timeoutInSeconds: 2,
  });
  if (!response.isExecuted) {
    console.log("Timeout or error:", response.error);
  }
} catch (err) {
  console.error("Command failed:", err.message);
}

HandleTimeout.java

try {
    CommandResponseMessage response = client.sendCommandRequest(
        CommandMessage.builder()
            .channel("orders.process")
            .body("test".getBytes())
            .timeout(2000)
            .build());
    if (!response.isExecuted()) {
        System.out.println("Timeout or error: " + response.getError());
    }
} catch (Exception e) {
    System.err.println("Command failed: " + e.getMessage());
}

HandleTimeout.cs

try
{
    var response = await client.SendCommandAsync(new CommandMessage
    {
        Channel = "orders.process",
        Body = Encoding.UTF8.GetBytes("test"),
        Timeout = TimeSpan.FromSeconds(2)
    });
    if (!response.IsExecuted)
        Console.WriteLine($"Timeout or error: {response.Error}");
}
catch (Exception ex)
{
    Console.WriteLine($"Command failed: {ex.Message}");
}

HandleTimeout.kt

try {
    val response = client.sendCommand(CommandMessage(
        channel = "orders.process",
        body = "test".toByteArray(),
        timeout = 2000
    ))
    if (!response.isExecuted) {
        println("Timeout or error: ${response.error}")
    }
} catch (e: Exception) {
    println("Command failed: ${e.message}")
}

handle_timeout.cpp

try {
    kubemq::CommandMessage cmd;
    cmd.channel = "orders.process";
    cmd.body = "test";
    cmd.timeout = 2000;

    auto response = client.sendCommand(cmd);
    if (!response.isExecuted) {
        std::cout << "Timeout or error: " << response.error << std::endl;
    }
} catch (const std::exception& e) {
    std::cerr << "Command failed: " << e.what() << std::endl;
}

handle_timeout.rs

use kubemq::CommandBuilder;
use std::time::Duration;

let command = CommandBuilder::new()
    .channel("orders.process")
    .body(b"test".to_vec())
    .timeout(Duration::from_secs(2))
    .build();

match client.send_command(command).await {
    Ok(resp) if !resp.executed => println!("Timeout or error: {}", resp.error),
    Ok(_) => println!("Command executed"),
    Err(e) => println!("Command failed: {}", e),
}

handle_timeout.rb

begin
  msg = KubeMQ::CQ::CommandMessage.new(
    channel: "orders.process",
    body: "test",
    timeout: 2_000 # milliseconds
  )
  response = client.send_command(msg)
  puts "Timeout or error: #{response.error}" unless response.executed
rescue KubeMQ::Error => e
  puts "Command failed: #{e.message}"
end

handle_timeout.exs

command = KubeMQ.Command.new(channel: "orders.process", body: "test", timeout: 2_000)

case KubeMQ.Client.send_command(client, command) do
  {:ok, response} ->
    unless response.executed, do: IO.puts("Timeout or error: #{response.error}")

  {:error, err} ->
    IO.puts("Command failed: #{err.message}")
end

Why Command Responses Are Stripped

KubeMQ follows the CQRS principle: Commands tell, they don't return data. When you send a command, the response body, metadata, and cacheHit fields are stripped by the server before being returned to the sender. Only Executed and Error reach the caller.

Response Field	Command	Query
`Body`	Stripped (always `nil`)	Preserved
`Metadata`	Stripped (always `""`)	Preserved
`CacheHit`	Stripped (always `false`)	Preserved

If you need to return data, use a Query instead.

Send Commands

What You Will Build

Steps

Create a Command Responder

Send a Command

Check Execution Status

Handle Timeout

Why Command Responses Are Stripped

Next Steps

Send Queries

Handle Commands

Configure Timeouts

On this page