Handle Command

Overview

A command handler is the receiving side of KubeMQ's Commands pattern — the code that actually does the work a caller is blocked waiting on. Instead of building your own request-routing layer on top of a queue, you register a handler once via the on_command callback passed to subscribe_to_commands, and KubeMQ delivers every matching command on that channel to it as a long-lived subscription, turning the channel into a synchronous RPC endpoint.

Handling happens inside on_command: you read the command's id, body, and reply_channel, run your business logic, then return a KubeMQ.CommandReply.new(request_id: cmd.id, response_to: cmd.reply_channel, executed: true, metadata: ...) — the SDK sends this reply automatically once the callback returns. Setting request_id and response_to to match the received command is what lets the broker correlate the reply back to the exact caller blocked on send_command; nothing else identifies which request the response belongs to.

Gotchas: the callback must return promptly and produce a valid CommandReply or the caller sees a timeout, since there's no separate explicit "send" call to retry; on_error only fires for transport-level subscription errors, not for exceptions raised inside on_command, so wrap risky logic yourself; and slow business logic inside the callback head-of-line blocks the next command on the same subscription.

Prerequisites

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

Code

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

{:ok, sub} =
  KubeMQ.Client.subscribe_to_commands(client, channel,
    on_command: fn cmd ->
      IO.puts("Processing command: #{cmd.body}")

      result = String.upcase(cmd.body)
      IO.puts("Result: #{result}")

      KubeMQ.CommandReply.new(
        request_id: cmd.id,
        response_to: cmd.reply_channel,
        executed: true,
        metadata: result
      )
    end,
    on_error: fn err ->
      IO.puts("Subscription error: #{err.message}")
    end
  )

IO.puts("Command handler ready on '#{channel}'")
Process.sleep(500)

cmd = KubeMQ.Command.new(channel: channel, body: "hello world", timeout: 10_000)

case KubeMQ.Client.send_command(client, cmd) do
  {:ok, resp} -> IO.puts("Response: executed=#{resp.executed}")
  {:error, err} -> IO.puts("Error: #{err.message}")
end

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

How It Works

• The on_command callback processes the command and returns a %CommandReply{}
• The reply's request_id and response_to must match the received command
• on_error is called if the subscription encounters transport errors

Related

• Send Command
• Command Group
• RPC Reference — struct field reference for Command/CommandReply