What You Will Build

A complete RPC flow: a responder that handles both commands and queries on the orders domain, and a sender that creates an order (command) then retrieves its status (query).

Steps

Set Up the Responder

The responder subscribes to both command and query channels and handles each type accordingly.

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("[CMD] Processing: %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("CMD error:", err) }),
    )
    if err != nil {
        log.Fatal(err)
    }

    _, err = client.SubscribeToQueries(ctx, "orders.lookup", "",
        kubemq.WithOnQueryReceive(func(query *kubemq.QueryReceive) {
            orderId := string(query.Body)
            fmt.Printf("[QRY] Looking up: %s\n", orderId)
            resp := kubemq.NewQueryReply().
                SetRequestId(query.Id).
                SetResponseTo(query.ResponseTo).
                SetBody([]byte(fmt.Sprintf(
                    `{"orderId":"%s","status":"confirmed","total":149.99}`, orderId))).
                SetExecutedAt(time.Now())
            _ = client.SendQueryResponse(ctx, resp)
        }),
        kubemq.WithOnError(func(err error) { log.Println("QRY error:", err) }),
    )
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println("Responder ready (commands + queries)...")
    <-ctx.Done()
}

responder.py

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

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

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

def on_query(request: QueryReceived) -> None:
    order_id = request.body.decode("utf-8")
    print(f"[QRY] Looking up: {order_id}")
    client.send_response_message(
        QueryResponse(
            query_received=request,
            is_executed=True,
            body=f'{{"orderId":"{order_id}","status":"confirmed","total":149.99}}'.encode(),
        )
    )

client.subscribe_to_commands(
    CommandsSubscription(channel="orders.process",
        on_receive_command_callback=on_command,
        on_error_callback=lambda e: print(f"CMD error: {e}")),
    cancel=cancel)

client.subscribe_to_queries(
    QueriesSubscription(channel="orders.lookup",
        on_receive_query_callback=on_query,
        on_error_callback=lambda e: print(f"QRY error: {e}")),
    cancel=cancel)

print("Responder ready (commands + queries)...")
time.sleep(3600)

responder.js

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

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

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

client.subscribeToQueries({
  channel: "orders.lookup",
  onQuery: (query) => {
    const orderId = Buffer.from(query.body).toString();
    console.log("[QRY] Looking up:", orderId);
    client.sendQueryResponse({
      requestId: query.id,
      isExecuted: true,
      body: Buffer.from(JSON.stringify({
        orderId, status: "confirmed", total: 149.99,
      })),
    });
  },
  onError: (err) => console.error("QRY error:", err.message),
});

console.log("Responder ready (commands + queries)...");

Responder.java

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

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

client.subscribeToQueries(QueriesSubscription.builder()
    .channel("orders.lookup")
    .onReceiveQueryCallback(query -> {
        String id = new String(query.getBody());
        System.out.println("[QRY] Looking up: " + id);
        return QueryResponseMessage.builder()
            .requestId(query.getId()).isExecuted(true)
            .body(String.format(
                "{\"orderId\":\"%s\",\"status\":\"confirmed\",\"total\":149.99}", id)
                .getBytes())
            .build();
    })
    .onErrorCallback(err -> System.err.println("QRY error: " + err.getMessage()))
    .build());

System.out.println("Responder ready (commands + queries)...");
Thread.sleep(3600000);
client.close();

Responder.cs

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

var cmdTask = Task.Run(async () =>
{
    await foreach (var cmd in client.SubscribeToCommandsAsync(
        new CommandsSubscription { Channel = "orders.process" }))
    {
        Console.WriteLine($"[CMD] Processing: {Encoding.UTF8.GetString(cmd.Body.Span)}");
        await client.SendCommandResponseAsync(new CommandResponse
        {
            RequestId = cmd.Id, IsExecuted = true
        });
    }
});

var queryTask = Task.Run(async () =>
{
    await foreach (var query in client.SubscribeToQueriesAsync(
        new QueriesSubscription { Channel = "orders.lookup" }))
    {
        var id = Encoding.UTF8.GetString(query.Body.Span);
        Console.WriteLine($"[QRY] Looking up: {id}");
        await client.SendQueryResponseAsync(new QueryResponse
        {
            RequestId = query.Id, IsExecuted = true,
            Body = Encoding.UTF8.GetBytes(
                $"{{\"orderId\":\"{id}\",\"status\":\"confirmed\",\"total\":149.99}}")
        });
    }
});

Console.WriteLine("Responder ready (commands + queries)...");
await Task.WhenAll(cmdTask, queryTask);

Responder.kt

val client = CQClient("localhost:50000")

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

client.subscribeToQueries(
    channel = "orders.lookup",
    onQuery = { query ->
        val id = String(query.body)
        println("[QRY] Looking up: $id")
        client.sendQueryResponse(
            requestId = query.id, isExecuted = true,
            body = """{"orderId":"$id","status":"confirmed","total":149.99}""".toByteArray()
        )
    },
    onError = { err -> System.err.println("QRY error: ${err.message}") }
)

println("Responder ready (commands + queries)...")
Thread.sleep(3600000)
client.close()

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 << "[CMD] Processing: " << cmd.body << std::endl;
        client.sendCommandResponse(cmd.id, true);
    },
    [](const std::string& err) { std::cerr << "CMD error: " << err << std::endl; }
);

client.subscribeToQueries("orders.lookup", "",
    [&client](const kubemq::QueryReceive& query) {
        std::cout << "[QRY] Looking up: " << query.body << std::endl;
        std::string data = R"({"orderId":")" + query.body +
            R"(","status":"confirmed","total":149.99})";
        client.sendQueryResponse(query.id, true, data);
    },
    [](const std::string& err) { std::cerr << "QRY error: " << err << std::endl; }
);

std::cout << "Responder ready (commands + queries)..." << std::endl;
std::this_thread::sleep_for(std::chrono::hours(1));

responder.rs

use kubemq::prelude::*;
use kubemq::{CommandReplyBuilder, QueryReplyBuilder};
use std::time::Duration;

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

    // Command responder — process orders, reply with executed only
    let cmd_client = client.clone();
    let _cmd_sub = client
        .subscribe_to_commands("orders.process", "", move |cmd| {
            let c = cmd_client.clone();
            Box::pin(async move {
                println!("[CMD] Processing: {}", String::from_utf8_lossy(&cmd.body));
                let reply = CommandReplyBuilder::new()
                    .request_id(&cmd.id)
                    .response_to(&cmd.response_to)
                    .executed_at(now_millis())
                    .build();
                tokio::spawn(async move { let _ = c.send_command_response(reply).await; });
            })
        }, None)
        .await?;

    // Query responder — look up orders, reply with a body payload
    let qry_client = client.clone();
    let _qry_sub = client
        .subscribe_to_queries("orders.lookup", "", move |query| {
            let c = qry_client.clone();
            Box::pin(async move {
                let order_id = String::from_utf8_lossy(&query.body).to_string();
                println!("[QRY] Looking up: {}", order_id);
                let body = format!(
                    r#"{{"orderId":"{}","status":"confirmed","total":149.99}}"#, order_id);
                let reply = QueryReplyBuilder::new()
                    .request_id(&query.id)
                    .response_to(&query.response_to)
                    .body(body.into_bytes())
                    .build();
                tokio::spawn(async move { let _ = c.send_query_response(reply).await; });
            })
        }, None)
        .await?;

    println!("Responder ready (commands + queries)...");
    tokio::time::sleep(Duration::from_secs(3600)).await;
    client.close().await?;
    Ok(())
}

fn now_millis() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as i64
}

responder.rb

require 'kubemq'

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

# Command responder — process orders, reply with executed only
cmd_sub = KubeMQ::CQ::CommandsSubscription.new(channel: 'orders.process')
client.subscribe_to_commands(cmd_sub, cancellation_token: cancel,
  on_error: ->(e) { puts "CMD error: #{e.message}" }) do |cmd|
  puts "[CMD] Processing: #{cmd.body}"
  client.send_response(KubeMQ::CQ::CommandResponseMessage.new(
    request_id: cmd.id, reply_channel: cmd.reply_channel, executed: true))
end

# Query responder — look up orders, reply with a body payload
qry_sub = KubeMQ::CQ::QueriesSubscription.new(channel: 'orders.lookup')
client.subscribe_to_queries(qry_sub, cancellation_token: cancel,
  on_error: ->(e) { puts "QRY error: #{e.message}" }) do |query|
  order_id = query.body
  puts "[QRY] Looking up: #{order_id}"
  client.send_response(KubeMQ::CQ::QueryResponseMessage.new(
    request_id: query.id, reply_channel: query.reply_channel, executed: true,
    body: "{\"orderId\":\"#{order_id}\",\"status\":\"confirmed\",\"total\":149.99}"))
end

puts 'Responder ready (commands + queries)...'
cancel.wait

responder.exs

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

# Command responder — process orders, reply with executed only
{:ok, _cmd_sub} =
  KubeMQ.Client.subscribe_to_commands(client, "orders.process",
    on_command: fn cmd ->
      IO.puts("[CMD] Processing: #{cmd.body}")

      KubeMQ.CommandReply.new(
        request_id: cmd.id,
        response_to: cmd.reply_channel,
        executed: true
      )
    end,
    on_error: fn err -> IO.puts("CMD error: #{err.message}") end
  )

# Query responder — look up orders, reply with a body payload
{:ok, _qry_sub} =
  KubeMQ.Client.subscribe_to_queries(client, "orders.lookup",
    on_query: fn query ->
      IO.puts("[QRY] Looking up: #{query.body}")

      KubeMQ.QueryReply.new(
        request_id: query.id,
        response_to: query.reply_channel,
        executed: true,
        body: ~s({"orderId":"#{query.body}","status":"confirmed","total":149.99})
      )
    end,
    on_error: fn err -> IO.puts("QRY error: #{err.message}") end
  )

IO.puts("Responder ready (commands + queries)...")
Process.sleep(3_600_000)
KubeMQ.Client.close(client)

Send a Command (Create Order)

In a separate terminal, send a command to create an order.

sender.go

cmdResp, err := client.SendCommand(ctx, kubemq.NewCommand().
    SetChannel("orders.process").
    SetBody([]byte(`{"action":"create","orderId":"ORD-9001"}`)).
    SetTimeout(10 * time.Second))
if err != nil {
    log.Fatal(err)
}
log.Printf("Command — Executed: %v", cmdResp.Executed)
log.Printf("Command — Body: %v (always nil)", cmdResp.Body)

sender.py

from kubemq.cq import Client as CQClient, CommandMessage, QueryMessage

with CQClient(address="localhost:50000") as client:
    cmd_resp = client.send_command(CommandMessage(
        channel="orders.process",
        body=b'{"action":"create","orderId":"ORD-9001"}',
        timeout_in_seconds=10))
    print(f"Command — Executed: {cmd_resp.is_executed}")
    print(f"Command — Body: {cmd_resp.body} (always empty)")

sender.js

const cmdResp = await client.sendCommand({
  channel: "orders.process",
  body: Buffer.from(JSON.stringify({ action: "create", orderId: "ORD-9001" })),
  timeoutInSeconds: 10,
});
console.log("Command — Executed:", cmdResp.isExecuted);
console.log("Command — Body:", cmdResp.body, "(always empty)");

Sender.java

var cmdResp = client.sendCommandRequest(CommandMessage.builder()
    .channel("orders.process")
    .body("{\"action\":\"create\",\"orderId\":\"ORD-9001\"}".getBytes())
    .timeout(10000).build());
System.out.println("Command — Executed: " + cmdResp.isExecuted());
System.out.println("Command — Body: " + cmdResp.getBody() + " (always null)");

Sender.cs

var cmdResp = await client.SendCommandAsync(new CommandMessage
{
    Channel = "orders.process",
    Body = Encoding.UTF8.GetBytes("{\"action\":\"create\",\"orderId\":\"ORD-9001\"}"),
    Timeout = TimeSpan.FromSeconds(10)
});
Console.WriteLine($"Command — Executed: {cmdResp.IsExecuted}");
Console.WriteLine($"Command — Body length: {cmdResp.Body.Length} (always 0)");

Sender.kt

val cmdResp = client.sendCommand(CommandMessage(
    channel = "orders.process",
    body = """{"action":"create","orderId":"ORD-9001"}""".toByteArray(),
    timeout = 10000))
println("Command — Executed: ${cmdResp.isExecuted}")
println("Command — Body: ${cmdResp.body} (always empty)")

sender.cpp

kubemq::CommandMessage cmd;
cmd.channel = "orders.process";
cmd.body = R"({"action":"create","orderId":"ORD-9001"})";
cmd.timeout = 10000;

auto cmdResp = client.sendCommand(cmd);
std::cout << "Command — Executed: " << cmdResp.isExecuted << std::endl;
std::cout << "Command — Body: " << cmdResp.body << " (always empty)" << std::endl;

sender.rs

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

let command = CommandBuilder::new()
    .channel("orders.process")
    .body(br#"{"action":"create","orderId":"ORD-9001"}"#.to_vec())
    .timeout(Duration::from_secs(10))
    .build();

let cmd_resp = client.send_command(command).await?;
println!("Command — Executed: {}", cmd_resp.executed);
println!("Command — Error: '{}' (no body on commands)", cmd_resp.error);

sender.rb

msg = KubeMQ::CQ::CommandMessage.new(
  channel: 'orders.process',
  timeout: 10,
  body: '{"action":"create","orderId":"ORD-9001"}'
)
cmd_resp = client.send_command(msg)
puts "Command — Executed: #{cmd_resp.executed}"
puts "Command — Error: #{cmd_resp.error} (no body on commands)"

sender.exs

command =
  KubeMQ.Command.new(
    channel: "orders.process",
    body: ~s({"action":"create","orderId":"ORD-9001"}),
    timeout: 10_000
  )

case KubeMQ.Client.send_command(client, command) do
  {:ok, cmd_resp} -> IO.puts("Command — Executed: #{cmd_resp.executed}")
  {:error, err} -> IO.puts("Command failed: #{err.message}")
end

Send a Query (Get Order Status)

Now query the order status. Unlike commands, the response body is preserved.

query.go

qryResp, err := client.SendQuery(ctx, kubemq.NewQuery().
    SetChannel("orders.lookup").
    SetBody([]byte("ORD-9001")).
    SetTimeout(10 * time.Second))
if err != nil {
    log.Fatal(err)
}
log.Printf("Query — Executed: %v", qryResp.Executed)
log.Printf("Query — Body: %s", qryResp.Body)

query.py

    qry_resp = client.send_query(QueryMessage(
        channel="orders.lookup",
        body=b"ORD-9001",
        timeout_in_seconds=10))
    print(f"Query — Executed: {qry_resp.is_executed}")
    print(f"Query — Body: {qry_resp.body.decode('utf-8')}")

query.js

const qryResp = await client.sendQuery({
  channel: "orders.lookup",
  body: Buffer.from("ORD-9001"),
  timeoutInSeconds: 10,
});
console.log("Query — Executed:", qryResp.isExecuted);
console.log("Query — Body:", Buffer.from(qryResp.body).toString());

Query.java

var qryResp = client.sendQueryRequest(QueryMessage.builder()
    .channel("orders.lookup")
    .body("ORD-9001".getBytes())
    .timeout(10000).build());
System.out.println("Query — Executed: " + qryResp.isExecuted());
System.out.println("Query — Body: " + new String(qryResp.getBody()));

Query.cs

var qryResp = await client.SendQueryAsync(new QueryMessage
{
    Channel = "orders.lookup",
    Body = Encoding.UTF8.GetBytes("ORD-9001"),
    Timeout = TimeSpan.FromSeconds(10)
});
Console.WriteLine($"Query — Executed: {qryResp.IsExecuted}");
Console.WriteLine($"Query — Body: {Encoding.UTF8.GetString(qryResp.Body.Span)}");

Query.kt

val qryResp = client.sendQuery(QueryMessage(
    channel = "orders.lookup",
    body = "ORD-9001".toByteArray(),
    timeout = 10000))
println("Query — Executed: ${qryResp.isExecuted}")
println("Query — Body: ${String(qryResp.body)}")

query.cpp

kubemq::QueryMessage query;
query.channel = "orders.lookup";
query.body = "ORD-9001";
query.timeout = 10000;

auto qryResp = client.sendQuery(query);
std::cout << "Query — Executed: " << qryResp.isExecuted << std::endl;
std::cout << "Query — Body: " << qryResp.body << std::endl;

query.rs

use kubemq::QueryBuilder;
use std::time::Duration;

let query = QueryBuilder::new()
    .channel("orders.lookup")
    .body(b"ORD-9001".to_vec())
    .timeout(Duration::from_secs(10))
    .build();

let qry_resp = client.send_query(query).await?;
println!("Query — Executed: {}", qry_resp.executed);
println!("Query — Body: {}", String::from_utf8_lossy(&qry_resp.body));

query.rb

msg = KubeMQ::CQ::QueryMessage.new(
  channel: 'orders.lookup',
  timeout: 10,
  body: 'ORD-9001'
)
qry_resp = client.send_query(msg)
puts "Query — Executed: #{qry_resp.executed}"
puts "Query — Body: #{qry_resp.body}"

query.exs

query =
  KubeMQ.Query.new(
    channel: "orders.lookup",
    body: "ORD-9001",
    timeout: 10_000
  )

case KubeMQ.Client.send_query(client, query) do
  {:ok, qry_resp} ->
    IO.puts("Query — Executed: #{qry_resp.executed}")
    IO.puts("Query — Body: #{qry_resp.body}")

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

Complete Flow

One responder, two request types: the command's response body is discarded, the query's response body is returned to the sender.

Key Takeaways

Commands for writes — response body is stripped, only Executed + Error returned
Queries for reads — response body is preserved, full data returned to sender
Same transport — both use the same client and same timeout mechanism
Responder must be running — if no responder is available, the sender gets a timeout error (code 301)