Request-Reply

Overview

Request-reply gives you synchronous RPC on top of KubeMQ's messaging fabric: a caller sends a query and blocks until the handler actually processing the request sends back a real answer — not just an acknowledgment. Reach for it whenever the caller needs a return value to proceed — a lookup, a computed result, a status check — the same shape as an HTTP call, but routed by KubeMQ instead of a service mesh or DNS.

A handler passed to subscribe_to_queries builds a QueryReplyBuilder with .request_id(&query.id) and .response_to(&query.response_to) before calling send_query_response — copying those fields is what lets KubeMQ route the response to the one caller waiting, not broadcast it. The caller's send_query awaits until that reply arrives or its .timeout(...) elapses, returning a response with a body.

Gotchas: if no subscriber is listening — or the handler crashes before replying — send_query simply times out; there's no way to distinguish "no handler" from "handler is slow" from the timeout alone. request_id and response_to must echo back the incoming query's values unchanged, or the reply is silently dropped or misrouted. If you don't actually need a return value, use commands instead — they only need an ack, so they don't tie up a caller waiting on a round trip.

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-patterns.request-reply";
    let rc = client.clone();

    let sub = client.subscribe_to_queries(channel, "",
        move |query| { let c = rc.clone(); Box::pin(async move {
            let result = format!("Processed: {}", String::from_utf8_lossy(&query.body));
            let reply = QueryReplyBuilder::new()
                .request_id(&query.id)
                .response_to(&query.response_to)
                .body(result.into_bytes())
                .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"get-user-profile".to_vec())
        .timeout(Duration::from_secs(10))
        .build();

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

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

How It Works

• The sender blocks until the handler returns a response or the timeout expires.
• Use Commands for fire-and-execute (no return data); use Queries for data retrieval.
• 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
• Configuration reference
• Fan-Out