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. 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
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

channel = "elixir-rpc.query-cached"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-query-cache")

call_count = :counters.new(1, [:atomics])

{:ok, sub} =
  KubeMQ.Client.subscribe_to_queries(client, channel,
    on_query: fn query ->
      :counters.add(call_count, 1, 1)
      IO.puts("[Handler] Processing query (call ##{:counters.get(call_count, 1)}): #{query.body}")

      KubeMQ.QueryReply.new(
        request_id: query.id,
        response_to: query.reply_channel,
        executed: true,
        body: "Result for #{query.body} at #{System.system_time(:second)}"
      )
    end
  )

Process.sleep(500)

query = KubeMQ.Query.new(
  channel: channel,
  body: "product-info",
  timeout: 10_000,
  cache_key: "product-info-cache",
  cache_ttl: 60_000
)

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

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

Process.sleep(500)

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

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

IO.puts("Handler was called #{:counters.get(call_count, 1)} time(s)")
KubeMQ.Subscription.cancel(sub)
KubeMQ.Client.close(client)

How It Works

• cache_key and cache_ttl enable server-side response caching
• The first query invokes the handler; the second returns the cached response
• response.cache_hit indicates whether the response came from cache
• The handler's :counters confirms it was only called once

Related

• Send Query
• Elixir SDK Reference