KubeMQ
LearnRPC

Getting Started with RPC

Send your first KubeMQ command and query in 5 minutes.

Prerequisites: KubeMQ server running on localhost:50000 and your SDK installed. See Getting Started for setup.

What You Will Build

A command sender that creates an order and a query sender that retrieves order status — each paired with a responder that handles the request and returns a response.

Steps

Install the SDK

go get github.com/kubemq-io/kubemq-go/v2
pip install kubemq
npm install kubemq-js
<dependency>
    <groupId>io.kubemq.sdk</groupId>
    <artifactId>kubemq-sdk-Java</artifactId>
    <version>2.1.1</version>
</dependency>
dotnet add package KubeMQ.SDK.CSharp
implementation("io.kubemq.sdk:kubemq-sdk-kotlin:2.1.0")
vcpkg install kubemq
[dependencies]
kubemq = "1.0"
tokio = { version = "1", features = ["full"] }
gem install kubemq
def deps do
  [{:kubemq, "~> 1.0"}]
end

Create a Command Responder

Start the responder first. It subscribes to incoming commands on orders.process, processes them, 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()

    sub, err := client.SubscribeToCommands(ctx, "orders.process", "",
        kubemq.WithOnCommandReceive(func(cmd *kubemq.CommandReceive) {
            fmt.Printf("Received command: %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)
    }
    defer sub.Unsubscribe()

    fmt.Println("Command responder listening 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"Received command: {request.body.decode('utf-8')}")
    client.send_response_message(
        CommandResponse(command_received=request, is_executed=True)
    )

def on_error(err):
    print(f"Error: {err}")

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=on_error,
    ),
    cancel=cancel,
)
print("Command responder listening 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("Received command:", Buffer.from(cmd.body).toString());
    client.sendCommandResponse({ requestId: cmd.id, isExecuted: true });
  },
  onError: (err) => console.error("Error:", err.message),
});

console.log("Command responder listening 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("Received: " + 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("Command responder listening on 'orders.process'...");
Thread.sleep(3600000);
client.close();
CommandResponder.cs
await using var client = new KubeMQClient(new KubeMQClientOptions());
await client.ConnectAsync();

Console.WriteLine("Command responder listening on 'orders.process'...");
await foreach (var cmd in client.SubscribeToCommandsAsync(
    new CommandsSubscription { Channel = "orders.process" }))
{
    Console.WriteLine($"Received: {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("Received command: ${String(cmd.body)}")
        client.sendCommandResponse(
            requestId = cmd.id, isExecuted = true
        )
    },
    onError = { err -> System.err.println("Error: ${err.message}") }
)

println("Command responder listening 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 << "Received command: " << cmd.body << std::endl;
        client.sendCommandResponse(cmd.id, true);
    },
    [](const std::string& err) {
        std::cerr << "Error: " << err << std::endl;
    }
);

std::cout << "Command responder listening on 'orders.process'..." << std::endl;
std::this_thread::sleep_for(std::chrono::hours(1));
command_responder.rs
use kubemq::prelude::*;
use kubemq::CommandReplyBuilder;
use std::time::Duration;

#[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!("Received command: {}", String::from_utf8_lossy(&cmd.body));
                    let reply = CommandReplyBuilder::new()
                        .request_id(&cmd.id)
                        .response_to(&cmd.response_to)
                        .build();
                    tokio::spawn(async move {
                        let _ = c.send_command_response(reply).await;
                    });
                })
            },
            None,
        )
        .await?;

    println!("Command responder listening on 'orders.process'...");
    tokio::signal::ctrl_c().await.ok();
    sub.unsubscribe().await;
    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 "Received command: #{cmd.body}"
  response = KubeMQ::CQ::CommandResponseMessage.new(
    request_id: cmd.id,
    reply_channel: cmd.reply_channel,
    executed: true
  )
  client.send_response(response)
end

puts "Command responder listening 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("Received command: #{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("Command responder listening on 'orders.process'...")
Process.sleep(:infinity)

Send a Command

In a separate terminal, send a command. The sender blocks until the responder replies or the timeout expires.

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

resp, err := client.SendCommand(ctx, kubemq.NewCommand().
    SetChannel("orders.process").
    SetBody([]byte(`{"action":"create","orderId":"ORD-1234"}`)).
    SetTimeout(10 * time.Second))
if err != nil {
    log.Fatal(err)
}
log.Printf("Executed: %v", resp.Executed)
send_command.py
from kubemq.cq import Client as CQClient
from kubemq.cq import CommandMessage

with CQClient(address="localhost:50000") as client:
    response = client.send_command(
        CommandMessage(
            channel="orders.process",
            body=b'{"action":"create","orderId":"ORD-1234"}',
            timeout_in_seconds=10,
        )
    )
    print(f"Executed: {response.is_executed}")
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-1234" })),
  timeoutInSeconds: 10,
});
console.log("Executed:", response.isExecuted);
SendCommand.java
CQClient client = CQClient.builder()
    .address("localhost:50000")
    .clientId("order-sender")
    .build();

CommandResponseMessage response = client.sendCommandRequest(
    CommandMessage.builder()
        .channel("orders.process")
        .body("{\"action\":\"create\",\"orderId\":\"ORD-1234\"}".getBytes())
        .timeout(10000)
        .build());
System.out.println("Executed: " + response.isExecuted());
client.close();
SendCommand.cs
await using var client = new KubeMQClient(new KubeMQClientOptions());
await client.ConnectAsync();

var response = await client.SendCommandAsync(new CommandMessage
{
    Channel = "orders.process",
    Body = Encoding.UTF8.GetBytes("{\"action\":\"create\",\"orderId\":\"ORD-1234\"}"),
    Timeout = TimeSpan.FromSeconds(10)
});
Console.WriteLine($"Executed: {response.IsExecuted}");
SendCommand.kt
val client = CQClient("localhost:50000")

val response = client.sendCommand(CommandMessage(
    channel = "orders.process",
    body = """{"action":"create","orderId":"ORD-1234"}""".toByteArray(),
    timeout = 10000
))
println("Executed: ${response.isExecuted}")
client.close()
send_command.cpp
#include <kubemq/client.h>
#include <iostream>

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

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

auto response = client.sendCommand(cmd);
std::cout << "Executed: " << response.isExecuted << std::endl;
send_command.rs
use kubemq::prelude::*;
use kubemq::CommandBuilder;
use std::time::Duration;

let client = KubemqClient::builder()
    .host("localhost")
    .port(50000)
    .build()
    .await?;

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

let response = client.send_command(command).await?;
println!("Executed: {}", response.executed);
client.close().await?;
send_command.rb
require 'kubemq'

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

msg = KubeMQ::CQ::CommandMessage.new(
  channel: 'orders.process',
  timeout: 10,
  body: '{"action":"create","orderId":"ORD-1234"}'
)
result = client.send_command(msg)
puts "Executed: #{result.executed}"
client.close
send_command.exs
{:ok, client} =
  KubeMQ.Client.start_link(address: "localhost:50000", client_id: "order-sender")

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

{:ok, response} = KubeMQ.Client.send_command(client, command)
IO.puts("Executed: #{response.executed}")
KubeMQ.Client.close(client)

Expected output (sender): Executed: true

Expected output (responder): Received command: {"action":"create","orderId":"ORD-1234"}

Create a Query Responder

Queries work like commands but the responder can include data in the response body. Set up a query responder on orders.lookup.

query_responder.go
sub, err := client.SubscribeToQueries(ctx, "orders.lookup", "",
    kubemq.WithOnQueryReceive(func(query *kubemq.QueryReceive) {
        fmt.Printf("Query for: %s\n", query.Body)
        resp := kubemq.NewQueryReply().
            SetRequestId(query.Id).
            SetResponseTo(query.ResponseTo).
            SetBody([]byte(`{"orderId":"ORD-1234","status":"shipped","total":99.99}`)).
            SetExecutedAt(time.Now())
        _ = client.SendQueryResponse(ctx, resp)
    }),
    kubemq.WithOnError(func(err error) {
        log.Println("Error:", err)
    }),
)
query_responder.py
from kubemq.cq import QueriesSubscription, QueryReceived, QueryResponse

def on_query(request: QueryReceived) -> None:
    print(f"Query for: {request.body.decode('utf-8')}")
    client.send_response_message(
        QueryResponse(
            query_received=request,
            is_executed=True,
            body=b'{"orderId":"ORD-1234","status":"shipped","total":99.99}',
        )
    )

client.subscribe_to_queries(
    subscription=QueriesSubscription(
        channel="orders.lookup",
        on_receive_query_callback=on_query,
        on_error_callback=on_error,
    ),
    cancel=cancel,
)
query_responder.js
client.subscribeToQueries({
  channel: "orders.lookup",
  onQuery: (query) => {
    console.log("Query for:", Buffer.from(query.body).toString());
    client.sendQueryResponse({
      requestId: query.id,
      isExecuted: true,
      body: Buffer.from(
        JSON.stringify({ orderId: "ORD-1234", status: "shipped", total: 99.99 })
      ),
    });
  },
  onError: (err) => console.error("Error:", err.message),
});
QueryResponder.java
client.subscribeToQueries(QueriesSubscription.builder()
    .channel("orders.lookup")
    .onReceiveQueryCallback(query -> {
        System.out.println("Query for: " + new String(query.getBody()));
        return QueryResponseMessage.builder()
            .requestId(query.getId())
            .isExecuted(true)
            .body("{\"orderId\":\"ORD-1234\",\"status\":\"shipped\",\"total\":99.99}".getBytes())
            .build();
    })
    .onErrorCallback(err -> System.err.println("Error: " + err.getMessage()))
    .build());
QueryResponder.cs
await foreach (var query in client.SubscribeToQueriesAsync(
    new QueriesSubscription { Channel = "orders.lookup" }))
{
    Console.WriteLine($"Query for: {Encoding.UTF8.GetString(query.Body.Span)}");
    await client.SendQueryResponseAsync(new QueryResponse
    {
        RequestId = query.Id,
        IsExecuted = true,
        Body = Encoding.UTF8.GetBytes(
            "{\"orderId\":\"ORD-1234\",\"status\":\"shipped\",\"total\":99.99}")
    });
}
QueryResponder.kt
client.subscribeToQueries(
    channel = "orders.lookup",
    onQuery = { query ->
        println("Query for: ${String(query.body)}")
        client.sendQueryResponse(
            requestId = query.id,
            isExecuted = true,
            body = """{"orderId":"ORD-1234","status":"shipped","total":99.99}""".toByteArray()
        )
    },
    onError = { err -> System.err.println("Error: ${err.message}") }
)
query_responder.cpp
client.subscribeToQueries("orders.lookup", "",
    [&client](const kubemq::QueryReceive& query) {
        std::cout << "Query for: " << query.body << std::endl;
        client.sendQueryResponse(query.id, true,
            R"({"orderId":"ORD-1234","status":"shipped","total":99.99})");
    },
    [](const std::string& err) {
        std::cerr << "Error: " << err << std::endl;
    }
);
query_responder.rs
use kubemq::QueryReplyBuilder;

let rc = client.clone();
let sub = client
    .subscribe_to_queries(
        "orders.lookup",
        "",
        move |query| {
            let c = rc.clone();
            Box::pin(async move {
                println!("Query for: {}", String::from_utf8_lossy(&query.body));
                let reply = QueryReplyBuilder::new()
                    .request_id(&query.id)
                    .response_to(&query.response_to)
                    .body(br#"{"orderId":"ORD-1234","status":"shipped","total":99.99}"#.to_vec())
                    .build();
                tokio::spawn(async move {
                    let _ = c.send_query_response(reply).await;
                });
            })
        },
        None,
    )
    .await?;
query_responder.rb
sub = KubeMQ::CQ::QueriesSubscription.new(channel: 'orders.lookup')
client.subscribe_to_queries(sub, cancellation_token: cancel,
  on_error: ->(e) { puts "Error: #{e.message}" }) do |query|
  puts "Query for: #{query.body}"
  response = KubeMQ::CQ::QueryResponseMessage.new(
    request_id: query.id,
    reply_channel: query.reply_channel,
    executed: true,
    body: '{"orderId":"ORD-1234","status":"shipped","total":99.99}'
  )
  client.send_response(response)
end
query_responder.exs
{:ok, _sub} =
  KubeMQ.Client.subscribe_to_queries(client, "orders.lookup",
    on_query: fn query ->
      IO.puts("Query for: #{query.body}")

      KubeMQ.QueryReply.new(
        request_id: query.id,
        response_to: query.reply_channel,
        executed: true,
        body: ~s({"orderId":"ORD-1234","status":"shipped","total":99.99})
      )
    end,
    on_error: fn err -> IO.puts("Error: #{err.message}") end
  )

Send a Query

Send a query to retrieve order data. Unlike commands, the response body is preserved.

send_query.go
resp, err := client.SendQuery(ctx, kubemq.NewQuery().
    SetChannel("orders.lookup").
    SetBody([]byte("ORD-1234")).
    SetTimeout(10 * time.Second))
if err != nil {
    log.Fatal(err)
}
log.Printf("Order data: %s", resp.Body)
send_query.py
from kubemq.cq import Client as CQClient
from kubemq.cq import QueryMessage

with CQClient(address="localhost:50000") as client:
    response = client.send_query(
        QueryMessage(
            channel="orders.lookup",
            body=b"ORD-1234",
            timeout_in_seconds=10,
        )
    )
    print(f"Order data: {response.body.decode('utf-8')}")
send_query.js
const response = await client.sendQuery({
  channel: "orders.lookup",
  body: Buffer.from("ORD-1234"),
  timeoutInSeconds: 10,
});
console.log("Order data:", Buffer.from(response.body).toString());
SendQuery.java
QueryResponseMessage response = client.sendQueryRequest(
    QueryMessage.builder()
        .channel("orders.lookup")
        .body("ORD-1234".getBytes())
        .timeout(10000)
        .build());
System.out.println("Order data: " + new String(response.getBody()));
SendQuery.cs
var response = await client.SendQueryAsync(new QueryMessage
{
    Channel = "orders.lookup",
    Body = Encoding.UTF8.GetBytes("ORD-1234"),
    Timeout = TimeSpan.FromSeconds(10)
});
Console.WriteLine($"Order data: {Encoding.UTF8.GetString(response.Body.Span)}");
SendQuery.kt
val response = client.sendQuery(QueryMessage(
    channel = "orders.lookup",
    body = "ORD-1234".toByteArray(),
    timeout = 10000
))
println("Order data: ${String(response.body)}")
send_query.cpp
kubemq::QueryMessage query;
query.channel = "orders.lookup";
query.body = "ORD-1234";
query.timeout = 10000;

auto response = client.sendQuery(query);
std::cout << "Order data: " << response.body << std::endl;
send_query.rs
use kubemq::QueryBuilder;
use std::time::Duration;

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

let response = client.send_query(query).await?;
println!("Order data: {}", String::from_utf8_lossy(&response.body));
send_query.rb
msg = KubeMQ::CQ::QueryMessage.new(
  channel: 'orders.lookup',
  timeout: 10,
  body: 'ORD-1234'
)
result = client.send_query(msg)
puts "Order data: #{result.body}"
send_query.exs
query =
  KubeMQ.Query.new(
    channel: "orders.lookup",
    body: "ORD-1234",
    timeout: 10_000
  )

{:ok, response} = KubeMQ.Client.send_query(client, query)
IO.puts("Order data: #{response.body}")

Unlike commands, query responses preserve the full response body and metadata. This makes queries ideal for data retrieval operations.

Run the Example

  1. Start the responder in one terminal (handles both commands and queries)
  2. Run the command sender in a separate terminal
  3. Run the query sender in a separate terminal
  4. Observe the responses in each terminal

What Just Happened

  1. The responder subscribed to both command and query channels
  2. The sender sent a command — the responder processed it and returned Executed: true (response body stripped)
  3. The sender sent a query — the responder returned the full order data in the response body (preserved)

Commands vs Queries Summary

AspectCommandQuery
Response bodyStripped (always nil)Preserved
Response metadataStrippedPreserved
CacheHit fieldStrippedPreserved
Use forWrites, mutations, actionsReads, lookups, data retrieval

Next Steps

Send Commands

Deep dive into command patterns with error handling.

Send Queries

Query patterns with response data handling.

Query Caching

Enable server-side response caching.

Reference

Complete request/response structure and error codes.

Was this page helpful?

RPC — Commands & Queries

Synchronous request-reply messaging with Commands for writes and Queries for reads, following CQRS principles.

Send Commands

Send fire-and-confirm commands with configurable timeouts and execution status.

On this page

What You Will BuildStepsInstall the SDKCreate a Command ResponderSend a CommandCreate a Query ResponderSend a QueryRun the ExampleWhat Just HappenedCommands vs Queries SummaryNext Steps