Handle Query

Overview

A query handler is the answering side of KubeMQ's request/response RPC pattern — the code that does real work and sends back data, unlike a Command handler, which only acknowledges receipt. Reach for it whenever a caller needs an actual answer — a lookup result, a computed value, a status object — not just confirmation that a message arrived.

Registering a handler with subscribe_to_queries and the on_query callback opens a subscription; the broker delivers every matching query to your function as it arrives. The callback builds a KubeMQ.QueryReply carrying the original query's correlation id (request_id/response_to) back to the broker, so the answer routes to the specific caller blocked waiting, and sets body with the real result before it's returned and sent.

Gotchas: if the handler never returns a reply, the caller blocks until its own timeout elapses and fails with an error, not a fast failure. An unhandled exception inside on_query doesn't automatically become a failure reply, so uncaught errors can leave the sender hanging. And because every matching query lands on the same callback, slow handler code delays every other in-flight caller.

Prerequisites

• KubeMQ server running on localhost:50000
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

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

{:ok, sub} =
  KubeMQ.Client.subscribe_to_queries(client, channel,
    on_query: fn query ->
      IO.puts("Calculating: #{query.body}")

      result =
        case query.body do
          "sum:" <> nums ->
            nums |> String.split(",") |> Enum.map(&String.to_integer(String.trim(&1))) |> Enum.sum()
          _ ->
            0
        end

      KubeMQ.QueryReply.new(
        request_id: query.id,
        response_to: query.reply_channel,
        executed: true,
        body: "#{result}"
      )
    end,
    on_error: fn err -> IO.puts("Error: #{err.message}") end
  )

IO.puts("Calculator query handler ready")
Process.sleep(500)

for input <- ["sum:1,2,3", "sum:10,20,30,40"] do
  query = KubeMQ.Query.new(channel: channel, body: input, timeout: 10_000)

  case KubeMQ.Client.send_query(client, query) do
    {:ok, resp} -> IO.puts("#{input} = #{resp.body}")
    {:error, err} -> IO.puts("Failed: #{err.message}")
  end
end

KubeMQ.Subscription.cancel(sub)
KubeMQ.Client.close(client)

How It Works

• The handler uses pattern matching on query.body to parse the input
• Results are computed and returned via %QueryReply{} with the body field
• Multiple queries are sent sequentially, each receiving a response

Related

• Send Query
• Cached Query
• RPC Reference — struct field reference for Query/QueryReply