Send Query

Overview

This tutorial builds the RPC half of KubeMQ's request/reply patterns: a query, where the caller awaits a handler's data payload instead of just a completion status. Reach for it whenever a caller needs an answer — fetching a record, running a lookup, or asking another service to compute a value on demand. You'll run a handler and a sender in the same process to see the full round trip.

The sender builds a request with QueryBuilder::new() and awaits client.send_query(query), which blocks until a reply arrives. client.subscribe_to_queries(...) registers an async handler closure; the handler builds a reply with QueryReplyBuilder::new().request_id(&query.id).response_to(&query.response_to) — copied from the incoming query — plus .body(...), sent with client.send_query_response(reply). KubeMQ routes that reply back to the caller waiting on it.

Gotchas: the .timeout(...) passed to the builder must cover however long the handler takes to run — a slow handler returns a timeout error even though the handler eventually succeeds. No handler subscribed yet also times out rather than erroring immediately. body is a raw Vec<u8> — encoding and decoding it (e.g. with String::from_utf8_lossy) is your application's job.

Prerequisites

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

Code

use kubemq::prelude::*;
use kubemq::{QueryBuilder, 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-send";
    let rc = client.clone();

    let sub = client.subscribe_to_queries(channel, "",
        move |query| { let c = rc.clone(); Box::pin(async move {
            println!("Query received: body={}", String::from_utf8_lossy(&query.body));
            let reply = QueryReplyBuilder::new()
                .request_id(&query.id)
                .response_to(&query.response_to)
                .body(b"query-result-data".to_vec())
                .build();
            let _ = c.send_query_response(reply).await;
        })}, None,
    ).await?;

    tokio::time::sleep(Duration::from_millis(500)).await;

    let query = QueryBuilder::new()
        .channel(channel)
        .body(b"lookup-request".to_vec())
        .timeout(Duration::from_secs(10))
        .build();

    let resp = client.send_query(query).await?;
    println!("Query response: executed={}, body={}", resp.executed, String::from_utf8_lossy(&resp.body));

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

How It Works

• Unlike commands, queries return response data in the body field.
• The handler uses QueryReplyBuilder to construct a response with payload.
• 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 Handle
• Query Cached