What You Will Build

An order lookup handler that receives queries, retrieves data, and returns structured responses in the response body. Unlike command responders, query responders send back full data payloads.

The handler subscribes, looks up the data, and returns it in the response body before the sender's timeout expires.

Steps

Subscribe to the orders.lookup channel. The handler receives each query, processes it, and must send a response before the sender's timeout expires.

query_handler.go

package main

import (
    "context"
    "encoding/json"
    "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) {
            handleQuery(ctx, client, query)
        }),
        kubemq.WithOnError(func(err error) {
            log.Println("Subscription error:", err)
        }),
    )
    if err != nil {
        log.Fatal(err)
    }

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

query_handler.py

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

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

client.subscribe_to_queries(
    subscription=QueriesSubscription(
        channel="orders.lookup",
        on_receive_query_callback=handle_query,
        on_error_callback=lambda e: print(f"Subscription error: {e}"),
    ),
    cancel=cancel,
)
print("Query handler ready on 'orders.lookup'...")
time.sleep(3600)

query_handler.js

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

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

client.subscribeToQueries({
  channel: "orders.lookup",
  onQuery: (query) => handleQuery(client, query),
  onError: (err) => console.error("Subscription error:", err.message),
});

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

QueryHandler.java

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

client.subscribeToQueries(QueriesSubscription.builder()
    .channel("orders.lookup")
    .onReceiveQueryCallback(query -> handleQuery(client, query))
    .onErrorCallback(err ->
        System.err.println("Subscription error: " + err.getMessage()))
    .build());

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

QueryHandler.cs

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

Console.WriteLine("Query handler ready on 'orders.lookup'...");
await foreach (var query in client.SubscribeToQueriesAsync(
    new QueriesSubscription { Channel = "orders.lookup" }))
{
    await HandleQuery(client, query);
}

QueryHandler.kt

val client = CQClient("localhost:50000")

client.subscribeToQueries(
    channel = "orders.lookup",
    onQuery = { query -> handleQuery(client, query) },
    onError = { err -> System.err.println("Subscription error: ${err.message}") }
)

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

query_handler.cpp

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

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

client.subscribeToQueries("orders.lookup", "",
    [&client](const kubemq::QueryReceive& query) {
        handleQuery(client, query);
    },
    [](const std::string& err) {
        std::cerr << "Subscription error: " << err << std::endl;
    }
);

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

query_handler.rs

use kubemq::prelude::*;
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 { handle_query(c, query).await })
            },
            None,
        )
        .await?;

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

query_handler.rb

require 'kubemq'

client = KubeMQ::CQClient.new(address: 'localhost:50000', client_id: 'order-query-handler')
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 "Subscription error: #{e.message}"
}) do |query|
  handle_query(client, query)
end

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

query_handler.exs

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

{:ok, _sub} =
  KubeMQ.Client.subscribe_to_queries(client, "orders.lookup",
    on_query: fn query -> handle_query(query) end,
    on_error: fn err -> IO.puts("Subscription error: #{err.message}") end
  )

IO.puts("Query handler ready on 'orders.lookup'...")
Process.sleep(:infinity)

Parse the Query

Read the request body to determine what data the caller is asking for.

parse.go

func handleQuery(ctx context.Context, client *kubemq.Client,
    query *kubemq.QueryReceive) {
    orderId := string(query.Body)
    fmt.Printf("Query for order: %s\n", orderId)

    order := lookupOrder(orderId)
    if order == nil {
        sendNotFound(ctx, client, query, orderId)
        return
    }

    data, _ := json.Marshal(order)
    resp := kubemq.NewQueryReply().
        SetRequestId(query.Id).
        SetResponseTo(query.ResponseTo).
        SetBody(data).
        SetMetadata("application/json").
        SetExecutedAt(time.Now())
    _ = client.SendQueryResponse(ctx, resp)
}

parse.py

def handle_query(request: QueryReceived) -> None:
    order_id = request.body.decode("utf-8")
    print(f"Query for order: {order_id}")

    order = lookup_order(order_id)
    if order is None:
        send_not_found(request, order_id)
        return

    client.send_response_message(
        QueryResponse(
            query_received=request,
            is_executed=True,
            body=json.dumps(order).encode(),
            metadata="application/json",
        )
    )

parse.js

function handleQuery(client, query) {
  const orderId = Buffer.from(query.body).toString();
  console.log("Query for order:", orderId);

  const order = lookupOrder(orderId);
  if (!order) {
    sendNotFound(client, query, orderId);
    return;
  }

  client.sendQueryResponse({
    id: query.id,
    replyChannel: query.replyChannel,
    executed: true,
    body: Buffer.from(JSON.stringify(order)),
    metadata: "application/json",
  });
}

Parse.java

private QueryResponseMessage handleQuery(CQClient client, QueryReceive query) {
    String orderId = new String(query.getBody());
    System.out.println("Query for order: " + orderId);

    Order order = lookupOrder(orderId);
    if (order == null) {
        return notFoundResponse(query, orderId);
    }

    return QueryResponseMessage.builder()
        .requestId(query.getId())
        .isExecuted(true)
        .body(new Gson().toJson(order).getBytes())
        .metadata("application/json")
        .build();
}

Parse.cs

async Task HandleQuery(KubeMQClient client, QueryReceive query)
{
    var orderId = Encoding.UTF8.GetString(query.Body.Span);
    Console.WriteLine($"Query for order: {orderId}");

    var order = LookupOrder(orderId);
    if (order is null)
    {
        await SendNotFound(client, query, orderId);
        return;
    }

    await client.SendQueryResponseAsync(new QueryResponse
    {
        RequestId = query.Id,
        IsExecuted = true,
        Body = JsonSerializer.SerializeToUtf8Bytes(order),
        Metadata = "application/json"
    });
}

Parse.kt

fun handleQuery(client: CQClient, query: QueryReceive) {
    val orderId = String(query.body)
    println("Query for order: $orderId")

    val order = lookupOrder(orderId)
    if (order == null) {
        sendNotFound(client, query, orderId)
        return
    }

    client.sendQueryResponse(
        requestId = query.id,
        isExecuted = true,
        body = Json.encodeToString(order).toByteArray(),
        metadata = "application/json"
    )
}

parse.cpp

void handleQuery(kubemq::CQClient& client,
    const kubemq::QueryReceive& query) {
    std::string orderId = query.body;
    std::cout << "Query for order: " << orderId << std::endl;

    auto order = lookupOrder(orderId);
    if (order.empty()) {
        sendNotFound(client, query, orderId);
        return;
    }

    client.sendQueryResponse(query.id, true, order, "application/json");
}

parse.rs

use kubemq::QueryReplyBuilder;

async fn handle_query(client: KubemqClient, query: QueryReceive) {
    let order_id = String::from_utf8_lossy(&query.body).to_string();
    println!("Query for order: {}", order_id);

    match lookup_order(&order_id) {
        Some(order) => {
            let reply = QueryReplyBuilder::new()
                .request_id(&query.id)
                .response_to(&query.response_to)
                .body(order.into_bytes())
                .metadata("application/json")
                .build();
            let _ = client.send_query_response(reply).await;
        }
        None => send_not_found(&client, &query, &order_id).await,
    }
}

parse.rb

def handle_query(client, query)
  order_id = query.body
  puts "Query for order: #{order_id}"

  order = lookup_order(order_id)
  if order.nil?
    send_not_found(client, query, order_id)
    return
  end

  response = KubeMQ::CQ::QueryResponseMessage.new(
    request_id: query.id,
    reply_channel: query.reply_channel,
    executed: true,
    body: order.to_json,
    metadata: 'application/json'
  )
  client.send_response(response)
end

parse.exs

def handle_query(query) do
  order_id = query.body
  IO.puts("Query for order: #{order_id}")

  case lookup_order(order_id) do
    nil ->
      send_not_found(query, order_id)

    order ->
      KubeMQ.QueryReply.new(
        request_id: query.id,
        response_to: query.reply_channel,
        executed: true,
        body: Jason.encode!(order),
        metadata: "application/json"
      )
  end
end

Return Data

Set the response body with the queried data. The body and metadata are preserved and returned to the sender.

return_data.go

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

func lookupOrder(id string) *Order {
    return &Order{
        OrderID: id,
        Status:  "shipped",
        Total:   149.99,
        Items:   3,
    }
}

return_data.py

def lookup_order(order_id: str) -> dict | None:
    return {
        "orderId": order_id,
        "status": "shipped",
        "total": 149.99,
        "items": 3,
    }

return_data.js

function lookupOrder(orderId) {
  return {
    orderId,
    status: "shipped",
    total: 149.99,
    items: 3,
  };
}

ReturnData.java

private Order lookupOrder(String id) {
    return new Order(id, "shipped", 149.99, 3);
}

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

ReturnData.cs

Order? LookupOrder(string id) =>
    new(id, "shipped", 149.99m, 3);

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

ReturnData.kt

fun lookupOrder(id: String): Order? =
    Order(orderId = id, status = "shipped", total = 149.99, items = 3)

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

return_data.cpp

std::string lookupOrder(const std::string& id) {
    nlohmann::json order;
    order["orderId"] = id;
    order["status"] = "shipped";
    order["total"] = 149.99;
    order["items"] = 3;
    return order.dump();
}

return_data.rs

use serde_json::json;

fn lookup_order(id: &str) -> Option<String> {
    Some(
        json!({
            "orderId": id,
            "status": "shipped",
            "total": 149.99,
            "items": 3
        })
        .to_string(),
    )
}

return_data.rb

require 'json'

def lookup_order(order_id)
  {
    orderId: order_id,
    status: 'shipped',
    total: 149.99,
    items: 3
  }
end

return_data.exs

def lookup_order(order_id) do
  %{
    orderId: order_id,
    status: "shipped",
    total: 149.99,
    items: 3
  }
end

Handle Not Found

Return Executed: false with an error message when the requested data does not exist.

not_found.go

func sendNotFound(ctx context.Context, client *kubemq.Client,
    query *kubemq.QueryReceive, orderId string) {
    resp := kubemq.NewQueryReply().
        SetRequestId(query.Id).
        SetResponseTo(query.ResponseTo).
        SetError(fmt.Sprintf("order %s not found", orderId))
    _ = client.SendQueryResponse(ctx, resp)
}

not_found.py

def send_not_found(request: QueryReceived, order_id: str) -> None:
    client.send_response_message(
        QueryResponse(
            query_received=request,
            is_executed=False,
            error=f"order {order_id} not found",
        )
    )

not_found.js

function sendNotFound(client, query, orderId) {
  client.sendQueryResponse({
    id: query.id,
    replyChannel: query.replyChannel,
    executed: false,
    error: `order ${orderId} not found`,
  });
}

NotFound.java

private QueryResponseMessage notFoundResponse(QueryReceive query, String id) {
    return QueryResponseMessage.builder()
        .requestId(query.getId())
        .isExecuted(false)
        .error("order " + id + " not found")
        .build();
}

NotFound.cs

async Task SendNotFound(KubeMQClient client, QueryReceive query, string id) =>
    await client.SendQueryResponseAsync(new QueryResponse
    {
        RequestId = query.Id,
        IsExecuted = false,
        Error = $"order {id} not found"
    });

NotFound.kt

fun sendNotFound(client: CQClient, query: QueryReceive, orderId: String) {
    client.sendQueryResponse(
        requestId = query.id,
        isExecuted = false,
        error = "order $orderId not found"
    )
}

not_found.cpp

void sendNotFound(kubemq::CQClient& client,
    const kubemq::QueryReceive& query, const std::string& orderId) {
    client.sendQueryResponse(query.id, false,
        "", "order " + orderId + " not found");
}

not_found.rs

use kubemq::QueryReplyBuilder;

async fn send_not_found(client: &KubemqClient, query: &QueryReceive, order_id: &str) {
    // Setting an error marks the reply as not executed.
    let reply = QueryReplyBuilder::new()
        .request_id(&query.id)
        .response_to(&query.response_to)
        .error(format!("order {} not found", order_id))
        .build();
    let _ = client.send_query_response(reply).await;
}

not_found.rb

def send_not_found(client, query, order_id)
  response = KubeMQ::CQ::QueryResponseMessage.new(
    request_id: query.id,
    reply_channel: query.reply_channel,
    executed: false,
    error: "order #{order_id} not found"
  )
  client.send_response(response)
end

not_found.exs

def send_not_found(query, order_id) do
  KubeMQ.QueryReply.new(
    request_id: query.id,
    response_to: query.reply_channel,
    executed: false,
    error: "order #{order_id} not found"
  )
end

Query vs Command Response

Response Field	Command Responder	Query Responder
`Body`	Not set (stripped by server)	Set response data
`Metadata`	Not set (stripped by server)	Set content type or metadata
`Executed`	Set success/failure	Set success/failure
`Error`	Set error message	Set error message

Handle Queries

What You Will Build

Steps

Parse the Query

Return Data

Handle Not Found

Query vs Command Response

Next Steps

Query Caching

Request-Reply Roundtrip

Reference

On this page