# Handle Queries (/learn/rpc/tutorials/handle-queries)



## What You Will Build [#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.

<Mermaid
  chart="sequenceDiagram
  participant C as Sender
  participant K as KubeMQ
  participant R as Query Handler
  C->>K: send query { orderId } (timeout)
  K->>R: deliver query
  R->>R: lookupOrder(orderId)
  R-->>K: response { executed, body: order }
  K-->>C: response { order data }"
/>

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

## Steps [#steps]

<Steps>
  <Step>
    ### Subscribe to Queries [#subscribe-to-queries]

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

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="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()
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="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)
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="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'...");
        ```
      </Tab>

      <Tab value="Java">
        ```java title="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();
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="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);
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="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()
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="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));
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="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(())
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="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
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="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)
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Parse the Query [#parse-the-query]

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

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="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)
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="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",
                )
            )
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="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",
          });
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="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();
        }
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="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"
            });
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="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"
            )
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="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");
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="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,
            }
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="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
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="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
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Return Data [#return-data]

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

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="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,
            }
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="return_data.py"
        def lookup_order(order_id: str) -> dict | None:
            return {
                "orderId": order_id,
                "status": "shipped",
                "total": 149.99,
                "items": 3,
            }
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="return_data.js"
        function lookupOrder(orderId) {
          return {
            orderId,
            status: "shipped",
            total: 149.99,
            items: 3,
          };
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="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) {}
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="ReturnData.cs"
        Order? LookupOrder(string id) =>
            new(id, "shipped", 149.99m, 3);

        record Order(string OrderId, string Status, decimal Total, int Items);
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="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)
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="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();
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="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(),
            )
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="return_data.rb"
        require 'json'

        def lookup_order(order_id)
          {
            orderId: order_id,
            status: 'shipped',
            total: 149.99,
            items: 3
          }
        end
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="return_data.exs"
        def lookup_order(order_id) do
          %{
            orderId: order_id,
            status: "shipped",
            total: 149.99,
            items: 3
          }
        end
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Handle Not Found [#handle-not-found]

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

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="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)
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="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",
                )
            )
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="not_found.js"
        function sendNotFound(client, query, orderId) {
          client.sendQueryResponse({
            id: query.id,
            replyChannel: query.replyChannel,
            executed: false,
            error: `order ${orderId} not found`,
          });
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="NotFound.java"
        private QueryResponseMessage notFoundResponse(QueryReceive query, String id) {
            return QueryResponseMessage.builder()
                .requestId(query.getId())
                .isExecuted(false)
                .error("order " + id + " not found")
                .build();
        }
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="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"
            });
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="NotFound.kt"
        fun sendNotFound(client: CQClient, query: QueryReceive, orderId: String) {
            client.sendQueryResponse(
                requestId = query.id,
                isExecuted = false,
                error = "order $orderId not found"
            )
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="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");
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="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;
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="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
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="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
        ```
      </Tab>
    </Tabs>
  </Step>
</Steps>

## Query vs Command Response [#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                |

## Next Steps [#next-steps]

<Cards>
  <Card title="Query Caching" href="/learn/rpc/tutorials/query-caching" description="Cache query responses server-side." />

  <Card title="Request-Reply Roundtrip" href="/learn/rpc/tutorials/request-reply-roundtrip" description="Full sender + responder in one example." />

  <Card title="Reference" href="/learn/rpc/reference" description="Complete response model and error codes." />
</Cards>