# Handle Commands (/learn/rpc/tutorials/handle-commands)



## What You Will Build [#what-you-will-build]

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

## How It Works [#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.

<Mermaid
  chart="sequenceDiagram
    participant S as Sender
    participant K as KubeMQ
    participant R as Responder
    S->>K: send command (orders.process)
    K->>R: deliver command
    R->>R: parse + execute business logic
    alt success
        R-->>K: Executed: true
    else failure
        R-->>K: Executed: false + error
    end
    K-->>S: command response"
/>

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

## Steps [#steps]

<Steps>
  <Step>
    ### Subscribe to Commands [#subscribe-to-commands]

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

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="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()
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="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)
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="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...");
        ```
      </Tab>

      <Tab value="Java">
        ```java title="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();
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="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);
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="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()
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="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));
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="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(())
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="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
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="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)
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Process the Command [#process-the-command]

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

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="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)
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="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)
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="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);
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="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");
            }
        }
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="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");
            }
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="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)
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="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);
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="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;
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="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
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="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
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Send Success Response [#send-success-response]

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

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="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)
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="success.py"
        def send_success(request: CommandReceived) -> None:
            client.send_response_message(
                CommandResponse(command_received=request, is_executed=True)
            )
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="success.js"
        function sendSuccess(client, cmd) {
          client.sendCommandResponse({ requestId: cmd.id, isExecuted: true });
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="Success.java"
        private CommandResponseMessage successResponse(CommandReceive cmd) {
            return CommandResponseMessage.builder()
                .requestId(cmd.getId())
                .isExecuted(true)
                .build();
        }
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="Success.cs"
        async Task SendSuccess(KubeMQClient client, CommandReceive cmd) =>
            await client.SendCommandResponseAsync(new CommandResponse
            {
                RequestId = cmd.Id, IsExecuted = true
            });
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="Success.kt"
        fun sendSuccess(client: CQClient, cmd: CommandReceive) {
            client.sendCommandResponse(requestId = cmd.id, isExecuted = true)
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="success.cpp"
        void sendSuccess(kubemq::CQClient& client,
            const kubemq::CommandReceive& cmd) {
            client.sendCommandResponse(cmd.id, true);
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="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;
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="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
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="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
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Send Error Response [#send-error-response]

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

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="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)
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="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,
                )
            )
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="error.js"
        function sendError(client, cmd, errorMsg) {
          client.sendCommandResponse({
            requestId: cmd.id,
            isExecuted: false,
            error: errorMsg,
          });
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="Error.java"
        private CommandResponseMessage errorResponse(CommandReceive cmd, String error) {
            return CommandResponseMessage.builder()
                .requestId(cmd.getId())
                .isExecuted(false)
                .error(error)
                .build();
        }
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="Error.cs"
        async Task SendError(KubeMQClient client, CommandReceive cmd, string error) =>
            await client.SendCommandResponseAsync(new CommandResponse
            {
                RequestId = cmd.Id, IsExecuted = false, Error = error
            });
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="Error.kt"
        fun sendError(client: CQClient, cmd: CommandReceive, errorMsg: String) {
            client.sendCommandResponse(
                requestId = cmd.id, isExecuted = false, error = errorMsg
            )
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="error.cpp"
        void sendError(kubemq::CQClient& client,
            const kubemq::CommandReceive& cmd, const std::string& errorMsg) {
            client.sendCommandResponse(cmd.id, false, errorMsg);
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="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;
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="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
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="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
        ```
      </Tab>
    </Tabs>
  </Step>
</Steps>

## Responder Best Practices [#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

## Next Steps [#next-steps]

<Cards>
  <Card title="Handle Queries" href="/learn/rpc/tutorials/handle-queries" description="Build a query responder that returns data." />

  <Card title="Load Balancing" href="/learn/rpc/how-to/load-balancing" description="Distribute commands across multiple responders." />

  <Card title="Request-Reply Roundtrip" href="/learn/rpc/tutorials/request-reply-roundtrip" description="Full sender + responder in one example." />
</Cards>