Cached Query

Overview

Query response caching lets the broker answer repeat requests without re-running your handler — useful when a query is expensive to compute (a database lookup, an aggregation, a downstream call) but the same input is asked for repeatedly in a short window. Only the first request pays the processing cost; every other caller gets the same answer straight from the broker.

Set cache_key and cache_ttl on the query builder. The first query with a given key is a miss: it reaches the handler, and the broker stores the response under that key for the TTL. A subsequent query with the same key is a hit — the broker returns the stored response directly without invoking the handler. cache_hit on the response tells you which happened.

Gotchas: the cache is keyed by the string you choose, not by the query body — if the underlying data changes mid-TTL, callers can get a stale answer until it expires. Keys are scoped per channel, so the same key on another channel is a separate entry. Caching only helps when requests genuinely repeat with the same key.

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

    let sub = client.subscribe_to_queries(channel, "",
        move |query| { let c = rc.clone(); Box::pin(async move {
            println!("Handler called (not cached)");
            let reply = QueryReplyBuilder::new()
                .request_id(&query.id)
                .response_to(&query.response_to)
                .body(b"expensive-result".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"sku-12345".to_vec())
        .timeout(Duration::from_secs(10))
        .cache_key("sku-12345")
        .cache_ttl(Duration::from_secs(60))
        .build();

    let resp1 = client.send_query(query.clone()).await?;
    println!("First call: cache_hit={}", resp1.cache_hit);

    let resp2 = client.send_query(query).await?;
    println!("Second call: cache_hit={}", resp2.cache_hit);

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

How It Works

• cache_key identifies the cached entry; cache_ttl sets how long responses are cached.
• The first call forwards to the handler; subsequent calls within the TTL return the cached result.
• resp.cache_hit indicates whether the response came from cache.
• 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