Handle Query

Overview

A query handler is the answering side of KubeMQ's request/response RPC pattern — the code that does real work and sends back data, unlike a Command handler, which only acknowledges receipt. Reach for it whenever a caller needs an actual answer — a lookup result, a computed value, a status object — not just confirmation that a message arrived.

Registering a handler with subscribe_to_queries opens a subscription; the broker delivers every matching query to your async closure as it arrives. The closure builds a reply with QueryReplyBuilder, carrying the original query's correlation id (request_id/response_to) back to the broker so the answer routes to the specific caller blocked waiting, and sets body with the real result before sending it with send_query_response.

Gotchas: if the handler never sends a response, the caller blocks until its own timeout elapses and fails with a timeout, not a fast error. A panic or error inside the closure doesn't automatically become a failure reply, so uncaught errors can leave the sender hanging. And because every matching query lands on the same closure, slow or blocking handler code delays every other in-flight caller.

Prerequisites

• KubeMQ server running on localhost:50000
• Rust SDK installed (cargo add kubemq)

Code

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 channel = "rust-rpc.query-handle";
    let rc = client.clone();

    let sub = client.subscribe_to_queries(channel, "",
        move |query| { let c = rc.clone(); Box::pin(async move {
            println!("Processing query: id={}", query.id);
            let reply = QueryReplyBuilder::new()
                .request_id(&query.id)
                .response_to(&query.response_to)
                .body(b"response-payload".to_vec())
                .metadata("application/json")
                .build();
            let _ = c.send_query_response(reply).await;
        })}, None,
    ).await?;

    println!("Listening for queries on '{}'...", channel);
    tokio::time::sleep(Duration::from_secs(30)).await;

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

How It Works

• QueryReply supports body, metadata, and error fields for rich responses.
• Setting .error("message") marks the query as not executed.
• Review timeouts, channel names, and client IDs before running against shared environments.
• Run the program while the server from the prerequisites is available.

Related

• RPC overview
• Rust SDK Reference
• Query Send