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 as on_query: to subscribe_to_queries builds a KubeMQ.QueryReply with request_id: query.id and response_to: query.reply_channel, returning it from the callback — copying those fields is what lets KubeMQ route the response to the one caller waiting, not broadcast it. The caller's send_query blocks until that reply arrives or its timeout elapses, returning {:ok, resp} with a body.

Gotchas: if no subscriber is listening — or the handler crashes before replying — send_query simply times out with {:error, err}; there's no way to distinguish "no handler" from "handler is slow" from the error 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
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

channel = "elixir-patterns.request-reply"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-req-reply")

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

      response_data =
        case query.body do
          "user:" <> id ->
            ~s({"id": "#{id}", "name": "Alice", "status": "active"})
          _ ->
            ~s({"error": "unknown request"})
        end

      KubeMQ.QueryReply.new(
        request_id: query.id,
        response_to: query.reply_channel,
        executed: true,
        body: response_data,
        metadata: "application/json"
      )
    end
  )

IO.puts("Service ready")
Process.sleep(500)

for user_id <- ["u-001", "u-002", "u-003"] do
  query = KubeMQ.Query.new(
    channel: channel,
    body: "user:#{user_id}",
    timeout: 10_000
  )

  case KubeMQ.Client.send_query(client, query) do
    {:ok, resp} ->
      IO.puts("Request user:#{user_id} → #{resp.body}")

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

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

How It Works

• The service subscribes to queries and returns user data based on the request body
• Pattern matching on query.body routes requests to the correct handler logic
• Multiple requests are sent sequentially, each receiving a JSON response

Related

• Send Query