# Send Commands (/learn/rpc/tutorials/send-commands)



## What You Will Build [#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.

<Mermaid
  chart="sequenceDiagram
    participant S as Sender
    participant K as KubeMQ
    participant R as Responder

    R->>K: subscribeToCommands(&#x22;orders.process&#x22;)
    S->>K: sendCommand(&#x22;orders.process&#x22;, body, timeout=10s)
    K->>R: deliver command
    R->>R: process order
    R->>K: response(Executed: true)
    K-->>S: deliver ack (Executed, Error — body stripped)"
/>

<p className="text-sm italic text-fd-muted-foreground">
  A command round-trip: the sender waits for an execution acknowledgment only — no data comes back.
</p>

## Steps [#steps]

<Steps>
  <Step>
    ### Create a Command Responder [#create-a-command-responder]

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

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

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

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

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

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

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

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

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

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

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

  <Step>
    ### Send a Command [#send-a-command]

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

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

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

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

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

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

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

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

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

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

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

  <Step>
    ### Check Execution Status [#check-execution-status]

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

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

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

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

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

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

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

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

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

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

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

  <Step>
    ### Handle Timeout [#handle-timeout]

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

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

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

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

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

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

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

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

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

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

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

## Why Command Responses Are Stripped [#why-command-responses-are-stripped]

KubeMQ follows the CQRS principle: &#x2A;*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](/learn/rpc/tutorials/send-queries) instead.

## Next Steps [#next-steps]

<Cards>
  <Card title="Send Queries" href="/learn/rpc/tutorials/send-queries" description="Send queries and receive structured response data." />

  <Card title="Handle Commands" href="/learn/rpc/tutorials/handle-commands" description="Build a command responder with error handling." />

  <Card title="Configure Timeouts" href="/learn/rpc/how-to/timeout-configuration" description="Set per-request timeouts and retry strategies." />
</Cards>