Send Command

Overview

A command is KubeMQ's fire-and-confirm RPC pattern: you reach for it when you need to know that an action actually ran on the other end — "process order #1234" — but you don't need any data back, just a yes/no on execution. It's the middle ground between one-way pub/sub, where you get no confirmation at all, and a query, where the handler returns a result payload. Commands turn "I hope that worked" into a definite outcome your caller can branch on.

This sample builds that lesson: KubeMQ.Client.subscribe_to_commands/3 registers an on_command callback, and KubeMQ.Client.send_command/2 blocks until a reply arrives or the timeout (in milliseconds) on KubeMQ.Command expires. The handler builds its reply as a %KubeMQ.CommandReply{} with request_id: cmd.id and response_to: cmd.reply_channel — that correlation is what lets the SDK route the response back to the exact caller waiting on it.

Gotchas: if no handler is subscribed (or it's still starting up), send_command/2 blocks for the full timeout before returning {:error, err} — there's no fast "nobody's listening" error. A handler that omits request_id/response_to on the reply leaves the caller hanging until timeout. And a command's reply carries no business data — if you need the handler to return a value, use a query instead.

Prerequisites

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

Code

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

{:ok, sub} =
  KubeMQ.Client.subscribe_to_commands(client, channel,
    on_command: fn cmd ->
      IO.puts("[Handler] Received command: #{cmd.body}")
      KubeMQ.CommandReply.new(
        request_id: cmd.id,
        response_to: cmd.reply_channel,
        executed: true
      )
    end
  )

Process.sleep(500)

command = KubeMQ.Command.new(
  channel: channel,
  body: "process order #1234",
  timeout: 10_000
)

case KubeMQ.Client.send_command(client, command) do
  {:ok, response} ->
    IO.puts("Command executed: #{response.executed}")
    IO.puts("Command ID: #{response.command_id}")

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

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

How It Works

• The on_command callback receives a %CommandReceive{} and returns a %CommandReply{}
• The SDK automatically sends the reply back to the caller
• send_command/2 blocks until a reply is received or the timeout expires
• The timeout field is required and specified in milliseconds

Related

• Handle Command
• Command Timeout
• RPC Pattern Overview