What You Will Build

A client that queries order status and receives full order data in the response. You will see how query responses differ from command responses — body, metadata, and cacheHit are all preserved.

A query round-trip: the responder returns structured data, and KubeMQ delivers the full response body and metadata back to the sender.

Steps

Create a Query Responder

The responder subscribes to the orders.lookup channel and returns order data in the response body.

query_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.SubscribeToQueries(ctx, "orders.lookup", "",
        kubemq.WithOnQueryReceive(func(query *kubemq.QueryReceive) {
            orderId := string(query.Body)
            fmt.Printf("Looking up order: %s\n", orderId)

            resp := kubemq.NewQueryReply().
                SetRequestId(query.Id).
                SetResponseTo(query.ResponseTo).
                SetBody([]byte(fmt.Sprintf(
                    `{"orderId":"%s","status":"shipped","total":99.99}`, orderId))).
                SetMetadata("application/json").
                SetExecutedAt(time.Now())
            _ = client.SendQueryResponse(ctx, resp)
        }),
        kubemq.WithOnError(func(err error) {
            log.Println("Error:", err)
        }),
    )
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println("Query responder ready on 'orders.lookup'...")
    <-ctx.Done()
}

query_responder.py

import time
from kubemq.cq import Client as CQClient
from kubemq.cq import QueriesSubscription, QueryReceived, QueryResponse, CancellationToken

def on_query(request: QueryReceived) -> None:
    order_id = request.body.decode("utf-8")
    print(f"Looking up order: {order_id}")
    client.send_response_message(
        QueryResponse(
            query_received=request,
            is_executed=True,
            body=f'{{"orderId":"{order_id}","status":"shipped","total":99.99}}'.encode(),
            metadata="application/json",
        )
    )

client = CQClient(address="localhost:50000")
cancel = CancellationToken()
client.subscribe_to_queries(
    subscription=QueriesSubscription(
        channel="orders.lookup",
        on_receive_query_callback=on_query,
        on_error_callback=lambda e: print(f"Error: {e}"),
    ),
    cancel=cancel,
)
print("Query responder ready on 'orders.lookup'...")
time.sleep(3600)

query_responder.js

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

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

client.subscribeToQueries({
  channel: "orders.lookup",
  onQuery: (query) => {
    const orderId = Buffer.from(query.body).toString();
    console.log("Looking up order:", orderId);
    client.sendQueryResponse({
      requestId: query.id,
      isExecuted: true,
      body: Buffer.from(
        JSON.stringify({ orderId, status: "shipped", total: 99.99 })
      ),
      metadata: "application/json",
    });
  },
  onError: (err) => console.error("Error:", err.message),
});

console.log("Query responder ready on 'orders.lookup'...");

QueryResponder.java

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

client.subscribeToQueries(QueriesSubscription.builder()
    .channel("orders.lookup")
    .onReceiveQueryCallback(query -> {
        String orderId = new String(query.getBody());
        System.out.println("Looking up order: " + orderId);
        return QueryResponseMessage.builder()
            .requestId(query.getId())
            .isExecuted(true)
            .body(String.format(
                "{\"orderId\":\"%s\",\"status\":\"shipped\",\"total\":99.99}",
                orderId).getBytes())
            .metadata("application/json")
            .build();
    })
    .onErrorCallback(err -> System.err.println("Error: " + err.getMessage()))
    .build());

System.out.println("Query responder ready on 'orders.lookup'...");
Thread.sleep(3600000);
client.close();

QueryResponder.cs

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

Console.WriteLine("Query responder ready on 'orders.lookup'...");
await foreach (var query in client.SubscribeToQueriesAsync(
    new QueriesSubscription { Channel = "orders.lookup" }))
{
    var orderId = Encoding.UTF8.GetString(query.Body.Span);
    Console.WriteLine($"Looking up order: {orderId}");
    await client.SendQueryResponseAsync(new QueryResponse
    {
        RequestId = query.Id,
        IsExecuted = true,
        Body = Encoding.UTF8.GetBytes(
            $"{{\"orderId\":\"{orderId}\",\"status\":\"shipped\",\"total\":99.99}}"),
        Metadata = "application/json"
    });
}

QueryResponder.kt

val client = CQClient("localhost:50000")

client.subscribeToQueries(
    channel = "orders.lookup",
    onQuery = { query ->
        val orderId = String(query.body)
        println("Looking up order: $orderId")
        client.sendQueryResponse(
            requestId = query.id,
            isExecuted = true,
            body = """{"orderId":"$orderId","status":"shipped","total":99.99}""".toByteArray(),
            metadata = "application/json"
        )
    },
    onError = { err -> System.err.println("Error: ${err.message}") }
)

println("Query responder ready on 'orders.lookup'...")
Thread.sleep(3600000)
client.close()

query_responder.cpp

#include <kubemq/client.h>
#include <iostream>
#include <thread>

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

client.subscribeToQueries("orders.lookup", "",
    [&client](const kubemq::QueryReceive& query) {
        std::cout << "Looking up order: " << query.body << std::endl;
        std::string response = R"({"orderId":")" + query.body +
            R"(","status":"shipped","total":99.99})";
        client.sendQueryResponse(query.id, true, response, "application/json");
    },
    [](const std::string& err) {
        std::cerr << "Error: " << err << std::endl;
    }
);

std::cout << "Query responder ready on 'orders.lookup'..." << std::endl;
std::this_thread::sleep_for(std::chrono::hours(1));

query_responder.rs

use kubemq::prelude::*;
use kubemq::QueryReplyBuilder;
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_queries(
            "orders.lookup",
            "",
            move |query| {
                let c = rc.clone();
                Box::pin(async move {
                    let order_id = String::from_utf8_lossy(&query.body).to_string();
                    println!("Looking up order: {}", order_id);
                    let reply = QueryReplyBuilder::new()
                        .request_id(&query.id)
                        .response_to(&query.response_to)
                        .body(
                            format!(
                                r#"{{"orderId":"{}","status":"shipped","total":99.99}}"#,
                                order_id
                            )
                            .into_bytes(),
                        )
                        .metadata("application/json")
                        .build();
                    tokio::spawn(async move {
                        let _ = c.send_query_response(reply).await;
                    });
                })
            },
            None,
        )
        .await?;

    println!("Query responder ready on 'orders.lookup'...");
    tokio::time::sleep(Duration::from_secs(3600)).await;
    Ok(())
}

query_responder.rb

require 'kubemq'

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

sub = KubeMQ::CQ::QueriesSubscription.new(channel: 'orders.lookup')
client.subscribe_to_queries(sub, cancellation_token: cancel, on_error: lambda { |e|
  puts "Error: #{e.message}"
}) do |query|
  order_id = query.body
  puts "Looking up order: #{order_id}"
  response = KubeMQ::CQ::QueryResponseMessage.new(
    request_id: query.id,
    reply_channel: query.reply_channel,
    executed: true,
    body: "{\"orderId\":\"#{order_id}\",\"status\":\"shipped\",\"total\":99.99}",
    metadata: 'application/json'
  )
  client.send_response(response)
end

puts "Query responder ready on 'orders.lookup'..."
cancel.wait

query_responder.exs

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

{:ok, _sub} =
  KubeMQ.Client.subscribe_to_queries(client, "orders.lookup",
    on_query: fn query ->
      IO.puts("Looking up order: #{query.body}")

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

IO.puts("Query responder ready on 'orders.lookup'...")
Process.sleep(3_600_000)

Send a Query

Send a query to retrieve order data. The response body and metadata are 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("Body: %s", resp.Body)
log.Printf("Metadata: %s", resp.Metadata)
log.Printf("Executed: %v", resp.Executed)

send_query.py

from kubemq.cq import Client as CQClient, 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"Body: {response.body.decode('utf-8')}")
    print(f"Metadata: {response.metadata}")
    print(f"Executed: {response.is_executed}")

send_query.js

const response = await client.sendQuery({
  channel: "orders.lookup",
  body: Buffer.from("ORD-1234"),
  timeoutInSeconds: 10,
});
console.log("Body:", Buffer.from(response.body).toString());
console.log("Metadata:", response.metadata);
console.log("Executed:", response.isExecuted);

SendQuery.java

QueryResponseMessage response = client.sendQueryRequest(
    QueryMessage.builder()
        .channel("orders.lookup")
        .body("ORD-1234".getBytes())
        .timeout(10000)
        .build());
System.out.println("Body: " + new String(response.getBody()));
System.out.println("Metadata: " + response.getMetadata());
System.out.println("Executed: " + response.isExecuted());

SendQuery.cs

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

SendQuery.kt

val response = client.sendQuery(QueryMessage(
    channel = "orders.lookup",
    body = "ORD-1234".toByteArray(),
    timeout = 10000
))
println("Body: ${String(response.body)}")
println("Metadata: ${response.metadata}")
println("Executed: ${response.isExecuted}")

send_query.cpp

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

auto response = client.sendQuery(query);
std::cout << "Body: " << response.body << std::endl;
std::cout << "Metadata: " << response.metadata << std::endl;
std::cout << "Executed: " << response.isExecuted << 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!("Body: {}", String::from_utf8_lossy(&response.body));
println!("Metadata: {}", response.metadata);
println!("Executed: {}", response.executed);

send_query.rb

msg = KubeMQ::CQ::QueryMessage.new(
  channel: 'orders.lookup',
  body: 'ORD-1234',
  timeout: 10
)

response = client.send_query(msg)
puts "Body: #{response.body}"
puts "Metadata: #{response.metadata}"
puts "Executed: #{response.executed}"

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("Body: #{response.body}")
IO.puts("Metadata: #{response.metadata}")
IO.puts("Executed: #{response.executed}")

Read Response Data

Parse the structured data returned in the query response body.

read_response.go

import "encoding/json"

type Order struct {
    OrderID string  `json:"orderId"`
    Status  string  `json:"status"`
    Total   float64 `json:"total"`
}

var order Order
if err := json.Unmarshal(resp.Body, &order); err != nil {
    log.Fatal(err)
}
log.Printf("Order %s: status=%s, total=$%.2f",
    order.OrderID, order.Status, order.Total)

read_response.py

import json

order = json.loads(response.body)
print(f"Order {order['orderId']}: status={order['status']}, total=${order['total']:.2f}")

read_response.js

const order = JSON.parse(Buffer.from(response.body).toString());
console.log(`Order ${order.orderId}: status=${order.status}, total=$${order.total}`);

ReadResponse.java

import com.google.gson.Gson;

record Order(String orderId, String status, double total) {}

Order order = new Gson().fromJson(new String(response.getBody()), Order.class);
System.out.printf("Order %s: status=%s, total=$%.2f%n",
    order.orderId(), order.status(), order.total());

ReadResponse.cs

using System.Text.Json;

var order = JsonSerializer.Deserialize<Order>(response.Body.Span);
Console.WriteLine($"Order {order.OrderId}: status={order.Status}, total=${order.Total:F2}");

record Order(string OrderId, string Status, decimal Total);

ReadResponse.kt

import kotlinx.serialization.json.Json

data class Order(val orderId: String, val status: String, val total: Double)

val order = Json.decodeFromString<Order>(String(response.body))
println("Order ${order.orderId}: status=${order.status}, total=$${order.total}")

read_response.cpp

#include <nlohmann/json.hpp>

auto order = nlohmann::json::parse(response.body);
std::cout << "Order " << order["orderId"]
          << ": status=" << order["status"]
          << ", total=$" << order["total"] << std::endl;

read_response.rs

use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    #[serde(rename = "orderId")]
    order_id: String,
    status: String,
    total: f64,
}

let order: Order = serde_json::from_slice(&response.body)?;
println!(
    "Order {}: status={}, total=${:.2}",
    order.order_id, order.status, order.total
);

read_response.rb

require 'json'

order = JSON.parse(response.body)
puts format('Order %s: status=%s, total=$%.2f',
            order['orderId'], order['status'], order['total'])

read_response.exs

order = Jason.decode!(response.body)

IO.puts(
  "Order #{order["orderId"]}: status=#{order["status"]}, " <>
    "total=$#{:erlang.float_to_binary(order["total"] / 1, decimals: 2)}"
)

Full Response Preserved

Unlike commands, query responses preserve all data fields:

Response Field	Command	Query
`Body`	Stripped (nil)	Preserved
`Metadata`	Stripped ("")	Preserved
`CacheHit`	Stripped (false)	Preserved
`Executed`	Preserved	Preserved
`Error`	Preserved	Preserved

Use queries for any operation that returns data. Use commands for write operations that need only an execution status.

Send Queries

What You Will Build

Steps

Create a Query Responder

Send a Query

Read Response Data

Full Response Preserved

Next Steps

Query Caching

Handle Queries

Reference

On this page