KubeMQ
LearnRPCTutorials

Query Caching

Enable server-side response caching for queries with configurable TTL.

What You Will Build

A cached product lookup — the first query hits the responder, and subsequent queries are served directly from KubeMQ's in-memory cache until the TTL expires.

How Query Caching Works

The first query misses the cache and reaches the responder; KubeMQ stores the reply and serves every later query with the same cache key directly until the TTL expires.

Steps

Set Up a Query Responder

The responder handles product lookups. With caching enabled, it is only called on cache misses.

product_responder.go
package main

import (
    "context"
    "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, "products.lookup", "",
        kubemq.WithOnQueryReceive(func(query *kubemq.QueryReceive) {
            productId := string(query.Body)
            fmt.Printf("Cache MISS — fetching product %s from database\n", productId)

            resp := kubemq.NewQueryReply().
                SetRequestId(query.Id).
                SetResponseTo(query.ResponseTo).
                SetBody([]byte(fmt.Sprintf(
                    `{"productId":"%s","name":"Widget","price":29.99,"stock":150}`,
                    productId))).
                SetExecutedAt(time.Now())
            _ = client.SendQueryResponse(ctx, resp)
        }),
        kubemq.WithOnError(func(err error) {
            log.Println("Error:", err)
        }),
    )
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println("Product responder ready...")
    <-ctx.Done()
}
product_responder.py
import time
from kubemq.cq import (
    Client as CQClient, QueriesSubscription,
    QueryReceived, QueryResponse, CancellationToken,
)

def on_query(request: QueryReceived) -> None:
    product_id = request.body.decode("utf-8")
    print(f"Cache MISS — fetching product {product_id} from database")
    client.send_response_message(
        QueryResponse(
            query_received=request,
            is_executed=True,
            body=f'{{"productId":"{product_id}","name":"Widget","price":29.99,"stock":150}}'.encode(),
        )
    )

client = CQClient(address="localhost:50000")
cancel = CancellationToken()
client.subscribe_to_queries(
    subscription=QueriesSubscription(
        channel="products.lookup",
        on_receive_query_callback=on_query,
        on_error_callback=lambda e: print(f"Error: {e}"),
    ),
    cancel=cancel,
)
print("Product responder ready...")
time.sleep(3600)
product_responder.js
const { KubeMQClient } = require("kubemq-js");

const client = new KubeMQClient({ address: "localhost:50000" });

client.subscribeToQueries({
  channel: "products.lookup",
  onQuery: (query) => {
    const productId = Buffer.from(query.body).toString();
    console.log(`Cache MISS — fetching product ${productId} from database`);
    client.sendQueryResponse({
      requestId: query.id,
      isExecuted: true,
      body: Buffer.from(JSON.stringify({
        productId, name: "Widget", price: 29.99, stock: 150,
      })),
    });
  },
  onError: (err) => console.error("Error:", err.message),
});

console.log("Product responder ready...");
ProductResponder.java
CQClient client = CQClient.builder()
    .address("localhost:50000")
    .clientId("product-responder")
    .build();

client.subscribeToQueries(QueriesSubscription.builder()
    .channel("products.lookup")
    .onReceiveQueryCallback(query -> {
        String productId = new String(query.getBody());
        System.out.println("Cache MISS — fetching product " + productId);
        return QueryResponseMessage.builder()
            .requestId(query.getId())
            .isExecuted(true)
            .body(String.format(
                "{\"productId\":\"%s\",\"name\":\"Widget\",\"price\":29.99,\"stock\":150}",
                productId).getBytes())
            .build();
    })
    .onErrorCallback(err -> System.err.println("Error: " + err.getMessage()))
    .build());

System.out.println("Product responder ready...");
Thread.sleep(3600000);
client.close();
ProductResponder.cs
await using var client = new KubeMQClient(new KubeMQClientOptions());
await client.ConnectAsync();

Console.WriteLine("Product responder ready...");
await foreach (var query in client.SubscribeToQueriesAsync(
    new QueriesSubscription { Channel = "products.lookup" }))
{
    var productId = Encoding.UTF8.GetString(query.Body.Span);
    Console.WriteLine($"Cache MISS — fetching product {productId}");
    await client.SendQueryResponseAsync(new QueryResponse
    {
        RequestId = query.Id,
        IsExecuted = true,
        Body = Encoding.UTF8.GetBytes(
            $"{{\"productId\":\"{productId}\",\"name\":\"Widget\",\"price\":29.99,\"stock\":150}}")
    });
}
ProductResponder.kt
val client = CQClient("localhost:50000")

client.subscribeToQueries(
    channel = "products.lookup",
    onQuery = { query ->
        val productId = String(query.body)
        println("Cache MISS — fetching product $productId from database")
        client.sendQueryResponse(
            requestId = query.id, isExecuted = true,
            body = """{"productId":"$productId","name":"Widget","price":29.99,"stock":150}""".toByteArray()
        )
    },
    onError = { err -> System.err.println("Error: ${err.message}") }
)

println("Product responder ready...")
Thread.sleep(3600000)
client.close()
product_responder.cpp
#include <kubemq/client.h>
#include <iostream>
#include <thread>

auto client = kubemq::CQClient("localhost:50000");

client.subscribeToQueries("products.lookup", "",
    [&client](const kubemq::QueryReceive& query) {
        std::cout << "Cache MISS — fetching product " << query.body << std::endl;
        std::string data = R"({"productId":")" + query.body +
            R"(","name":"Widget","price":29.99,"stock":150})";
        client.sendQueryResponse(query.id, true, data);
    },
    [](const std::string& err) {
        std::cerr << "Error: " << err << std::endl;
    }
);

std::cout << "Product responder ready..." << std::endl;
std::this_thread::sleep_for(std::chrono::hours(1));
product_responder.rs
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 responder = client.clone();
    let sub = client
        .subscribe_to_queries(
            "products.lookup",
            "",
            move |query| {
                let rc = responder.clone();
                Box::pin(async move {
                    let product_id = String::from_utf8_lossy(&query.body).to_string();
                    println!("Cache MISS — fetching product {product_id} from database");

                    let reply = QueryReplyBuilder::new()
                        .request_id(&query.id)
                        .response_to(&query.response_to)
                        .body(
                            format!(
                                r#"{{"productId":"{product_id}","name":"Widget","price":29.99,"stock":150}}"#
                            )
                            .into_bytes(),
                        )
                        .build();
                    tokio::spawn(async move {
                        let _ = rc.send_query_response(reply).await;
                    });
                })
            },
            None,
        )
        .await?;

    println!("Product responder ready...");
    tokio::signal::ctrl_c().await.ok();
    sub.unsubscribe().await;
    client.close().await
}
product_responder.rb
require 'kubemq'

client = KubeMQ::CQClient.new(address: 'localhost:50000', client_id: 'product-responder')
cancel = KubeMQ::CancellationToken.new

sub = KubeMQ::CQ::QueriesSubscription.new(channel: 'products.lookup')
client.subscribe_to_queries(sub, cancellation_token: cancel, on_error: lambda { |e|
  puts "Error: #{e.message}"
}) do |query|
  product_id = query.body
  puts "Cache MISS — fetching product #{product_id} from database"
  response = KubeMQ::CQ::QueryResponseMessage.new(
    request_id: query.id,
    reply_channel: query.reply_channel,
    executed: true,
    body: "{\"productId\":\"#{product_id}\",\"name\":\"Widget\",\"price\":29.99,\"stock\":150}"
  )
  client.send_response(response)
end

puts 'Product responder ready...'
cancel.wait
product_responder.exs
{:ok, client} =
  KubeMQ.Client.start_link(address: "localhost:50000", client_id: "product-responder")

{:ok, _sub} =
  KubeMQ.Client.subscribe_to_queries(client, "products.lookup",
    on_query: fn query ->
      product_id = query.body
      IO.puts("Cache MISS — fetching product #{product_id} from database")

      KubeMQ.QueryReply.new(
        request_id: query.id,
        response_to: query.reply_channel,
        executed: true,
        body: ~s({"productId":"#{product_id}","name":"Widget","price":29.99,"stock":150})
      )
    end,
    on_error: fn err -> IO.puts("Error: #{err.message}") end
  )

IO.puts("Product responder ready...")
Process.sleep(:infinity)

Send a Query with CacheKey

Set CacheKey and CacheTTL on the query request to enable caching. The first query is a cache miss, and subsequent queries within the TTL are served from cache.

cached_query.go
resp, err := client.SendQuery(ctx, kubemq.NewQuery().
    SetChannel("products.lookup").
    SetBody([]byte("PROD-123")).
    SetTimeout(10 * time.Second).
    SetCacheKey("product-PROD-123").
    SetCacheTTL(60 * time.Second))
if err != nil {
    log.Fatal(err)
}
log.Printf("Product: %s (cache hit: %v)", resp.Body, resp.CacheHit)
cached_query.py
from kubemq.cq import Client as CQClient, QueryMessage

with CQClient(address="localhost:50000") as client:
    response = client.send_query(
        QueryMessage(
            channel="products.lookup",
            body=b"PROD-123",
            timeout_in_seconds=10,
            cache_key="product-PROD-123",
            cache_ttl_in_seconds=60,
        )
    )
    print(f"Product: {response.body.decode('utf-8')} (cache hit: {response.cache_hit})")
cached_query.js
const response = await client.sendQuery({
  channel: "products.lookup",
  body: Buffer.from("PROD-123"),
  timeoutInSeconds: 10,
  cacheKey: "product-PROD-123",
  cacheTTL: 60000,
});
console.log("Product:", Buffer.from(response.body).toString(),
  "(cache hit:", response.cacheHit, ")");
CachedQuery.java
QueryResponseMessage response = client.sendQueryRequest(
    QueryMessage.builder()
        .channel("products.lookup")
        .body("PROD-123".getBytes())
        .timeout(10000)
        .cacheKey("product-PROD-123")
        .cacheTTL(60000)
        .build());
System.out.println("Product: " + new String(response.getBody())
    + " (cache hit: " + response.isCacheHit() + ")");
CachedQuery.cs
var response = await client.SendQueryAsync(new QueryMessage
{
    Channel = "products.lookup",
    Body = Encoding.UTF8.GetBytes("PROD-123"),
    Timeout = TimeSpan.FromSeconds(10),
    CacheKey = "product-PROD-123",
    CacheTTL = TimeSpan.FromSeconds(60)
});
Console.WriteLine($"Product: {Encoding.UTF8.GetString(response.Body.Span)}"
    + $" (cache hit: {response.CacheHit})");
CachedQuery.kt
val response = client.sendQuery(QueryMessage(
    channel = "products.lookup",
    body = "PROD-123".toByteArray(),
    timeout = 10000,
    cacheKey = "product-PROD-123",
    cacheTTL = 60000
))
println("Product: ${String(response.body)} (cache hit: ${response.cacheHit})")
cached_query.cpp
kubemq::QueryMessage query;
query.channel = "products.lookup";
query.body = "PROD-123";
query.timeout = 10000;
query.cacheKey = "product-PROD-123";
query.cacheTTL = 60000;

auto response = client.sendQuery(query);
std::cout << "Product: " << response.body
          << " (cache hit: " << response.cacheHit << ")" << std::endl;
cached_query.rs
use kubemq::QueryBuilder;
use std::time::Duration;

let query = QueryBuilder::new()
    .channel("products.lookup")
    .body(b"PROD-123".to_vec())
    .timeout(Duration::from_secs(10))
    .cache_key("product-PROD-123")
    .cache_ttl(Duration::from_secs(60))
    .build();

let response = client.send_query(query).await?;
println!(
    "Product: {} (cache hit: {})",
    String::from_utf8_lossy(&response.body),
    response.cache_hit
);
cached_query.rb
msg = KubeMQ::CQ::QueryMessage.new(
  channel: 'products.lookup',
  body: 'PROD-123',
  timeout: 10,
  cache_key: 'product-PROD-123',
  cache_ttl: 60
)

response = client.send_query(msg)
puts "Product: #{response.body} (cache hit: #{response.cache_hit})"
cached_query.exs
query =
  KubeMQ.Query.new(
    channel: "products.lookup",
    body: "PROD-123",
    timeout: 10_000,
    cache_key: "product-PROD-123",
    cache_ttl: 60_000
  )

case KubeMQ.Client.send_query(client, query) do
  {:ok, resp} ->
    IO.puts("Product: #{resp.body} (cache hit: #{resp.cache_hit})")

  {:error, err} ->
    IO.puts("Error: #{err.message}")
end

Verify Cache Behavior

Send the same query twice to observe the cache in action.

verify_cache.go
query := kubemq.NewQuery().
    SetChannel("products.lookup").
    SetBody([]byte("PROD-123")).
    SetTimeout(10 * time.Second).
    SetCacheKey("product-PROD-123").
    SetCacheTTL(60 * time.Second)

resp1, _ := client.SendQuery(ctx, query)
log.Printf("First query  — cache hit: %v", resp1.CacheHit) // false

resp2, _ := client.SendQuery(ctx, query)
log.Printf("Second query — cache hit: %v", resp2.CacheHit) // true
verify_cache.py
query = QueryMessage(
    channel="products.lookup",
    body=b"PROD-123",
    timeout_in_seconds=10,
    cache_key="product-PROD-123",
    cache_ttl_in_seconds=60,
)

resp1 = client.send_query(query)
print(f"First query  — cache hit: {resp1.cache_hit}")  # False

resp2 = client.send_query(query)
print(f"Second query — cache hit: {resp2.cache_hit}")  # True
verify_cache.js
const queryOpts = {
  channel: "products.lookup",
  body: Buffer.from("PROD-123"),
  timeoutInSeconds: 10,
  cacheKey: "product-PROD-123",
  cacheTTL: 60000,
};

const resp1 = await client.sendQuery(queryOpts);
console.log("First query  — cache hit:", resp1.cacheHit); // false

const resp2 = await client.sendQuery(queryOpts);
console.log("Second query — cache hit:", resp2.cacheHit); // true
VerifyCache.java
QueryMessage query = QueryMessage.builder()
    .channel("products.lookup")
    .body("PROD-123".getBytes())
    .timeout(10000)
    .cacheKey("product-PROD-123")
    .cacheTTL(60000)
    .build();

var resp1 = client.sendQueryRequest(query);
System.out.println("First query  — cache hit: " + resp1.isCacheHit()); // false

var resp2 = client.sendQueryRequest(query);
System.out.println("Second query — cache hit: " + resp2.isCacheHit()); // true
VerifyCache.cs
var query = new QueryMessage
{
    Channel = "products.lookup",
    Body = Encoding.UTF8.GetBytes("PROD-123"),
    Timeout = TimeSpan.FromSeconds(10),
    CacheKey = "product-PROD-123",
    CacheTTL = TimeSpan.FromSeconds(60)
};

var resp1 = await client.SendQueryAsync(query);
Console.WriteLine($"First query  — cache hit: {resp1.CacheHit}"); // false

var resp2 = await client.SendQueryAsync(query);
Console.WriteLine($"Second query — cache hit: {resp2.CacheHit}"); // true
VerifyCache.kt
val query = QueryMessage(
    channel = "products.lookup",
    body = "PROD-123".toByteArray(),
    timeout = 10000,
    cacheKey = "product-PROD-123",
    cacheTTL = 60000
)

val resp1 = client.sendQuery(query)
println("First query  — cache hit: ${resp1.cacheHit}") // false

val resp2 = client.sendQuery(query)
println("Second query — cache hit: ${resp2.cacheHit}") // true
verify_cache.cpp
kubemq::QueryMessage query;
query.channel = "products.lookup";
query.body = "PROD-123";
query.timeout = 10000;
query.cacheKey = "product-PROD-123";
query.cacheTTL = 60000;

auto resp1 = client.sendQuery(query);
std::cout << "First query  — cache hit: " << resp1.cacheHit << std::endl; // 0

auto resp2 = client.sendQuery(query);
std::cout << "Second query — cache hit: " << resp2.cacheHit << std::endl; // 1
verify_cache.rs
use kubemq::QueryBuilder;
use std::time::Duration;

let build_query = || {
    QueryBuilder::new()
        .channel("products.lookup")
        .body(b"PROD-123".to_vec())
        .timeout(Duration::from_secs(10))
        .cache_key("product-PROD-123")
        .cache_ttl(Duration::from_secs(60))
        .build()
};

let resp1 = client.send_query(build_query()).await?;
println!("First query  — cache hit: {}", resp1.cache_hit); // false

let resp2 = client.send_query(build_query()).await?;
println!("Second query — cache hit: {}", resp2.cache_hit); // true
verify_cache.rb
msg = KubeMQ::CQ::QueryMessage.new(
  channel: 'products.lookup',
  body: 'PROD-123',
  timeout: 10,
  cache_key: 'product-PROD-123',
  cache_ttl: 60
)

resp1 = client.send_query(msg)
puts "First query  — cache hit: #{resp1.cache_hit}"  # false

resp2 = client.send_query(msg)
puts "Second query — cache hit: #{resp2.cache_hit}"  # true
verify_cache.exs
query =
  KubeMQ.Query.new(
    channel: "products.lookup",
    body: "PROD-123",
    timeout: 10_000,
    cache_key: "product-PROD-123",
    cache_ttl: 60_000
  )

{:ok, resp1} = KubeMQ.Client.send_query(client, query)
IO.puts("First query  — cache hit: #{resp1.cache_hit}")  # false

{:ok, resp2} = KubeMQ.Client.send_query(client, query)
IO.puts("Second query — cache hit: #{resp2.cache_hit}")  # true

Cache Configuration

SettingDescription
CacheKeyAny string identifying the cached response (use entity-type-id pattern)
CacheTTLTime-to-live in milliseconds. Required when CacheKey is set.
StorageIn-memory on the KubeMQ server
Cleanup intervalEvery 10 seconds
PersistenceNone — cache is cleared on server restart

Caching is only available for Queries. Commands do not support caching because they represent write operations.

When to Use Caching

Use CaseCache?Why
Product catalog lookups✅ YesData changes infrequently
Configuration values✅ YesRarely updated
Exchange rates✅ YesUpdate every few minutes
User-specific data⚠️ DependsUse unique cache keys per user
Real-time stock prices❌ NoData is stale immediately
Authentication tokens❌ NoSecurity-sensitive

Next Steps

Request-Reply Roundtrip

Complete sender + responder example.

Reference

Full caching configuration details.

Was this page helpful?

Handle Queries

Build a query responder that processes incoming requests and returns structured data.

Request-Reply Roundtrip

Implement the full request-reply cycle showing both sender and responder in a single example.

On this page

What You Will BuildHow Query Caching WorksStepsSet Up a Query ResponderSend a Query with CacheKeyVerify Cache BehaviorCache ConfigurationWhen to Use CachingNext Steps