Think of RPC like a phone call — you dial a number, ask a question, and wait on the line for an answer. If nobody picks up within a set time, you hang up and try again. KubeMQ RPC brings this synchronous request-reply model to messaging infrastructure.

KubeMQ implements RPC through two complementary operation types following the CQRS (Command Query Responsibility Segregation) principle: Commands for writes and Queries for reads.

The concept it implements. RPC is KubeMQ's realization of the request/reply interaction style — the sender blocks on a single matched response. The timeout-and-retry behavior maps to the delivery guarantees you choose at the application level: a request either gets exactly one answer or a timeout error.

Commands vs Queries

Aspect	Commands	Queries
Purpose	State-changing operations (writes, mutations)	Read-only operations (data lookups)
Response body	Stripped — sender receives only execution status	Preserved — sender receives full response payload
Response metadata	Stripped	Preserved
Caching	Not supported	Supported via `CacheKey` / `CacheTTL`
Use cases	Order placement, device control, config changes	Data lookups, status checks, service reads

Commands tell the system to do something and return only a success/failure indicator. Queries ask for data and return the full response body and metadata.

How It Works

Request/reply: the sender blocks while KubeMQ routes the request to a responder and delivers the single matched response back.

The sender publishes a request to a named channel with a timeout. KubeMQ routes the request to a subscribed responder (or load-balances across a group of responders). The responder processes the request and sends a response back through KubeMQ. If no response arrives before the timeout expires, the sender receives a timeout error.

Key Features

Synchronous request-reply — sender blocks until a response arrives or timeout expires
Commands and Queries — separate semantics for writes and reads following CQRS
Response caching — server-side caching for queries with configurable TTL
Load balancing — distribute requests across multiple responders using queue groups
Configurable timeouts — per-request timeout in milliseconds
gRPC and REST — use any transport protocol

Quick Example

send_command.go

package main

import (
    "context"
    "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()

    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("Command executed: %v", resp.Executed)
}

send_command.py

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

client = CQClient(address="localhost:50000")
response = client.send_command(
    CommandMessage(
        channel="orders.process",
        body=b'{"action":"create","orderId":"ORD-1234"}',
        timeout_in_seconds=10,
    )
)
print(f"Command executed: {response.is_executed}")
client.close()

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("Command executed:", response.isExecuted);

SendCommand.java

CQClient client = CQClient.builder()
    .address("localhost:50000")
    .clientId("order-service")
    .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("Command 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 << "Command executed: " << response.isExecuted << std::endl;

send_command.rs

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

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    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!("Command executed: {}", response.executed);

    client.close().await?;
    Ok(())
}

send_command.rb

require 'kubemq'

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

msg = KubeMQ::CQ::CommandMessage.new(
  channel: 'orders.process',
  body: '{"action":"create","orderId":"ORD-1234"}',
  timeout: 10
)
result = client.send_command(msg)
puts "Command executed: #{result.executed}"

client.close

send_command.exs

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

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

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

KubeMQ.Client.close(client)

When to Use RPC

Scenario	RPC	Events / Queues
Service-to-service API calls	✅ Best choice	Not suitable
Data lookups and reads	✅ Queries	Possible but awkward
Write confirmations	✅ Commands	Queues with ack
CQRS implementation	✅ Commands + Queries	Events for projections
Fire-and-forget broadcasts	❌ Blocks on response	✅ Use Events
Reliable async processing	❌ Blocks on response	✅ Use Queues

Need fire-and-forget delivery? Use Events for broadcasts or Queues for reliable processing.

Commands and queries are also available via the CloudEvents protocol — use any language with a CloudEvents SDK, no KubeMQ client library needed.

RPC — Commands & Queries

Commands vs Queries

How It Works

Key Features

Quick Example

When to Use RPC

Learn More

Getting Started

Send Commands

Send Queries

Handle Commands

Query Caching

Load Balancing

Reference

On this page