What You Will Build

An order processing handler that receives commands, executes business logic, and sends back success or failure responses.

How It Works

A responder subscribes to a command channel, processes each incoming command, and replies with an execution result. KubeMQ correlates the reply back to the blocked sender.

Responder flow: KubeMQ routes each command to the handler and returns its execution result to the waiting sender.

Steps

Subscribe to the orders.process channel to receive incoming commands. Optionally specify a group name for load balancing across multiple responders.

subscribe.go

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"

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

type OrderCommand struct {
    Action  string `json:"action"`
    OrderID string `json:"orderId"`
}

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) {
            handleCommand(ctx, client, cmd)
        }),
        kubemq.WithOnError(func(err error) {
            log.Println("Subscription error:", err)
        }),
    )
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println("Command handler ready...")
    <-ctx.Done()
}

subscribe.py

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

client = CQClient(address="localhost:50000")
cancel = CancellationToken()

def handle_command(request: CommandReceived) -> None:
    order = json.loads(request.body)
    print(f"Handling {order['action']} for {order['orderId']}")

    # Process the command (see next step)
    client.send_response_message(
        CommandResponse(command_received=request, is_executed=True)
    )

client.subscribe_to_commands(
    subscription=CommandsSubscription(
        channel="orders.process",
        on_receive_command_callback=handle_command,
        on_error_callback=lambda e: print(f"Subscription error: {e}"),
    ),
    cancel=cancel,
)
print("Command handler ready...")
time.sleep(3600)

subscribe.js

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

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

client.subscribeToCommands({
  channel: "orders.process",
  onCommand: (cmd) => handleCommand(client, cmd),
  onError: (err) => console.error("Subscription error:", err.message),
});

console.log("Command handler ready...");

Subscribe.java

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

client.subscribeToCommands(CommandsSubscription.builder()
    .channel("orders.process")
    .onReceiveCommandCallback(cmd -> handleCommand(client, cmd))
    .onErrorCallback(err ->
        System.err.println("Subscription error: " + err.getMessage()))
    .build());

System.out.println("Command handler ready...");
Thread.sleep(3600000);
client.close();

Subscribe.cs

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

Console.WriteLine("Command handler ready...");
await foreach (var cmd in client.SubscribeToCommandsAsync(
    new CommandsSubscription { Channel = "orders.process" }))
{
    await HandleCommand(client, cmd);
}

Subscribe.kt

val client = CQClient("localhost:50000")

client.subscribeToCommands(
    channel = "orders.process",
    onCommand = { cmd -> handleCommand(client, cmd) },
    onError = { err -> System.err.println("Subscription error: ${err.message}") }
)

println("Command handler ready...")
Thread.sleep(3600000)
client.close()

subscribe.cpp

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

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

client.subscribeToCommands("orders.process", "",
    [&client](const kubemq::CommandReceive& cmd) {
        handleCommand(client, cmd);
    },
    [](const std::string& err) {
        std::cerr << "Subscription error: " << err << std::endl;
    }
);

std::cout << "Command handler ready..." << std::endl;
std::this_thread::sleep_for(std::chrono::hours(1));

subscribe.rs

use kubemq::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct OrderCommand {
    action: String,
    #[serde(rename = "orderId")]
    order_id: String,
}

#[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",
            "", // group: "" for no load balancing
            move |cmd| {
                let c = rc.clone();
                Box::pin(async move { handle_command(c, cmd).await })
            },
            None,
        )
        .await?;

    println!("Command handler ready...");
    tokio::signal::ctrl_c().await.ok();
    client.close().await?;
    Ok(())
}

subscribe.rb

require 'kubemq'

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

sub = KubeMQ::CQ::CommandsSubscription.new(channel: 'orders.process')
client.subscribe_to_commands(
  sub,
  cancellation_token: cancel,
  on_error: ->(e) { puts "Subscription error: #{e.message}" }
) do |cmd|
  handle_command(client, cmd)
end

puts 'Command handler ready...'
cancel.wait

subscribe.exs

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

# The on_command callback returns a CommandReply — the SDK sends it automatically
{:ok, _sub} =
  KubeMQ.Client.subscribe_to_commands(client, "orders.process",
    on_command: fn cmd -> handle_command(cmd) end,
    on_error: fn err -> IO.puts("Subscription error: #{err.message}") end
  )

IO.puts("Command handler ready...")
Process.sleep(:infinity)

Process the Command

Parse the request body, execute business logic, and determine the result.

process.go

func handleCommand(ctx context.Context, client *kubemq.Client,
    cmd *kubemq.CommandReceive) {

    var order OrderCommand
    if err := json.Unmarshal(cmd.Body, &order); err != nil {
        sendError(ctx, client, cmd, "invalid request body")
        return
    }

    switch order.Action {
    case "create":
        fmt.Printf("Creating order %s\n", order.OrderID)
    case "cancel":
        fmt.Printf("Cancelling order %s\n", order.OrderID)
    default:
        sendError(ctx, client, cmd, "unknown action: "+order.Action)
        return
    }

    sendSuccess(ctx, client, cmd)
}

process.py

def handle_command(request: CommandReceived) -> None:
    try:
        order = json.loads(request.body)
    except json.JSONDecodeError:
        send_error(request, "invalid request body")
        return

    action = order.get("action")
    if action == "create":
        print(f"Creating order {order['orderId']}")
    elif action == "cancel":
        print(f"Cancelling order {order['orderId']}")
    else:
        send_error(request, f"unknown action: {action}")
        return

    send_success(request)

process.js

function handleCommand(client, cmd) {
  let order;
  try {
    order = JSON.parse(Buffer.from(cmd.body).toString());
  } catch {
    sendError(client, cmd, "invalid request body");
    return;
  }

  switch (order.action) {
    case "create":
      console.log(`Creating order ${order.orderId}`);
      break;
    case "cancel":
      console.log(`Cancelling order ${order.orderId}`);
      break;
    default:
      sendError(client, cmd, `unknown action: ${order.action}`);
      return;
  }

  sendSuccess(client, cmd);
}

Process.java

private CommandResponseMessage handleCommand(CQClient client,
    CommandReceive cmd) {
    try {
        var order = new Gson().fromJson(
            new String(cmd.getBody()), OrderCommand.class);

        switch (order.action()) {
            case "create" -> System.out.println("Creating order " + order.orderId());
            case "cancel" -> System.out.println("Cancelling order " + order.orderId());
            default -> {
                return errorResponse(cmd, "unknown action: " + order.action());
            }
        }
        return successResponse(cmd);
    } catch (Exception e) {
        return errorResponse(cmd, "invalid request body");
    }
}

Process.cs

async Task HandleCommand(KubeMQClient client, CommandReceive cmd)
{
    try
    {
        var order = JsonSerializer.Deserialize<OrderCommand>(cmd.Body.Span);

        switch (order?.Action)
        {
            case "create": Console.WriteLine($"Creating order {order.OrderId}"); break;
            case "cancel": Console.WriteLine($"Cancelling order {order.OrderId}"); break;
            default: await SendError(client, cmd, $"unknown action: {order?.Action}"); return;
        }
        await SendSuccess(client, cmd);
    }
    catch
    {
        await SendError(client, cmd, "invalid request body");
    }
}

Process.kt

fun handleCommand(client: CQClient, cmd: CommandReceive) {
    val order = try {
        Json.decodeFromString<OrderCommand>(String(cmd.body))
    } catch (e: Exception) {
        sendError(client, cmd, "invalid request body")
        return
    }

    when (order.action) {
        "create" -> println("Creating order ${order.orderId}")
        "cancel" -> println("Cancelling order ${order.orderId}")
        else -> { sendError(client, cmd, "unknown action: ${order.action}"); return }
    }
    sendSuccess(client, cmd)
}

process.cpp

void handleCommand(kubemq::CQClient& client,
    const kubemq::CommandReceive& cmd) {
    auto order = nlohmann::json::parse(cmd.body, nullptr, false);
    if (order.is_discarded()) {
        sendError(client, cmd, "invalid request body");
        return;
    }

    auto action = order["action"].get<std::string>();
    if (action == "create") {
        std::cout << "Creating order " << order["orderId"] << std::endl;
    } else if (action == "cancel") {
        std::cout << "Cancelling order " << order["orderId"] << std::endl;
    } else {
        sendError(client, cmd, "unknown action: " + action);
        return;
    }
    sendSuccess(client, cmd);
}

process.rs

async fn handle_command(client: KubemqClient, cmd: CommandReceive) {
    let order: OrderCommand = match serde_json::from_slice(&cmd.body) {
        Ok(o) => o,
        Err(_) => return send_error(client, &cmd, "invalid request body").await,
    };

    match order.action.as_str() {
        "create" => println!("Creating order {}", order.order_id),
        "cancel" => println!("Cancelling order {}", order.order_id),
        other => return send_error(client, &cmd, &format!("unknown action: {other}")).await,
    }

    send_success(client, &cmd).await;
}

process.rb

def handle_command(client, cmd)
  order = begin
    JSON.parse(cmd.body)
  rescue JSON::ParserError
    return send_error(client, cmd, 'invalid request body')
  end

  case order['action']
  when 'create' then puts "Creating order #{order['orderId']}"
  when 'cancel' then puts "Cancelling order #{order['orderId']}"
  else return send_error(client, cmd, "unknown action: #{order['action']}")
  end

  send_success(client, cmd)
end

process.exs

def handle_command(cmd) do
  case Jason.decode(cmd.body) do
    {:ok, %{"action" => "create", "orderId" => id}} ->
      IO.puts("Creating order #{id}")
      send_success(cmd)

    {:ok, %{"action" => "cancel", "orderId" => id}} ->
      IO.puts("Cancelling order #{id}")
      send_success(cmd)

    {:ok, %{"action" => action}} ->
      send_error(cmd, "unknown action: #{action}")

    {:error, _} ->
      send_error(cmd, "invalid request body")
  end
end

Send Success Response

Return Executed: true to indicate the command was processed successfully.

success.go

func sendSuccess(ctx context.Context, client *kubemq.Client,
    cmd *kubemq.CommandReceive) {
    resp := kubemq.NewCommandReply().
        SetRequestId(cmd.Id).
        SetResponseTo(cmd.ResponseTo).
        SetExecutedAt(time.Now())
    _ = client.SendCommandResponse(ctx, resp)
}

success.py

def send_success(request: CommandReceived) -> None:
    client.send_response_message(
        CommandResponse(command_received=request, is_executed=True)
    )

success.js

function sendSuccess(client, cmd) {
  client.sendCommandResponse({ requestId: cmd.id, isExecuted: true });
}

Success.java

private CommandResponseMessage successResponse(CommandReceive cmd) {
    return CommandResponseMessage.builder()
        .requestId(cmd.getId())
        .isExecuted(true)
        .build();
}

Success.cs

async Task SendSuccess(KubeMQClient client, CommandReceive cmd) =>
    await client.SendCommandResponseAsync(new CommandResponse
    {
        RequestId = cmd.Id, IsExecuted = true
    });

Success.kt

fun sendSuccess(client: CQClient, cmd: CommandReceive) {
    client.sendCommandResponse(requestId = cmd.id, isExecuted = true)
}

success.cpp

void sendSuccess(kubemq::CQClient& client,
    const kubemq::CommandReceive& cmd) {
    client.sendCommandResponse(cmd.id, true);
}

success.rs

async fn send_success(client: KubemqClient, cmd: &CommandReceive) {
    // A reply with no error means Executed: true
    let reply = CommandReplyBuilder::new()
        .request_id(&cmd.id)
        .response_to(&cmd.response_to)
        .build();
    let _ = client.send_command_response(reply).await;
}

success.rb

def send_success(client, cmd)
  response = KubeMQ::CQ::CommandResponseMessage.new(
    request_id: cmd.id,
    reply_channel: cmd.reply_channel,
    executed: true
  )
  client.send_response(response)
end

success.exs

# The callback returns a CommandReply — the SDK sends it automatically
def send_success(cmd) do
  KubeMQ.CommandReply.new(
    request_id: cmd.id,
    response_to: cmd.reply_channel,
    executed: true
  )
end

Send Error Response

Return Executed: false with an error message when processing fails.

error.go

func sendError(ctx context.Context, client *kubemq.Client,
    cmd *kubemq.CommandReceive, errMsg string) {
    resp := kubemq.NewCommandReply().
        SetRequestId(cmd.Id).
        SetResponseTo(cmd.ResponseTo).
        SetError(errMsg)
    _ = client.SendCommandResponse(ctx, resp)
}

error.py

def send_error(request: CommandReceived, error_msg: str) -> None:
    client.send_response_message(
        CommandResponse(
            command_received=request,
            is_executed=False,
            error=error_msg,
        )
    )

error.js

function sendError(client, cmd, errorMsg) {
  client.sendCommandResponse({
    requestId: cmd.id,
    isExecuted: false,
    error: errorMsg,
  });
}

Error.java

private CommandResponseMessage errorResponse(CommandReceive cmd, String error) {
    return CommandResponseMessage.builder()
        .requestId(cmd.getId())
        .isExecuted(false)
        .error(error)
        .build();
}

Error.cs

async Task SendError(KubeMQClient client, CommandReceive cmd, string error) =>
    await client.SendCommandResponseAsync(new CommandResponse
    {
        RequestId = cmd.Id, IsExecuted = false, Error = error
    });

Error.kt

fun sendError(client: CQClient, cmd: CommandReceive, errorMsg: String) {
    client.sendCommandResponse(
        requestId = cmd.id, isExecuted = false, error = errorMsg
    )
}

error.cpp

void sendError(kubemq::CQClient& client,
    const kubemq::CommandReceive& cmd, const std::string& errorMsg) {
    client.sendCommandResponse(cmd.id, false, errorMsg);
}

error.rs

async fn send_error(client: KubemqClient, cmd: &CommandReceive, err_msg: &str) {
    // Setting an error marks the reply as not executed
    let reply = CommandReplyBuilder::new()
        .request_id(&cmd.id)
        .response_to(&cmd.response_to)
        .error(err_msg)
        .build();
    let _ = client.send_command_response(reply).await;
}

error.rb

def send_error(client, cmd, error_msg)
  response = KubeMQ::CQ::CommandResponseMessage.new(
    request_id: cmd.id,
    reply_channel: cmd.reply_channel,
    executed: false,
    error: error_msg
  )
  client.send_response(response)
end

error.exs

# Return a CommandReply with executed: false — the SDK sends it automatically
def send_error(cmd, error_msg) do
  KubeMQ.CommandReply.new(
    request_id: cmd.id,
    response_to: cmd.reply_channel,
    executed: false,
    error: error_msg
  )
end

Responder Best Practices

Keep processing fast — the sender is blocking and waiting for your response
Always send a response — if you don't respond, the sender will timeout (code 301)
Use groups for scaling — multiple responders with the same group name share load via round-robin
Handle unknown commands gracefully — return Executed: false with a descriptive error

Handle Commands

What You Will Build

How It Works

Steps

Process the Command

Send Success Response

Send Error Response

Responder Best Practices

Next Steps

Handle Queries

Load Balancing

Request-Reply Roundtrip

On this page