Command Group

Overview

A command consumer group turns a single command handler into a scalable worker pool: run multiple identical handlers subscribed with the same group name, and the broker load-balances each incoming command to exactly one member instead of broadcasting it to all of them. This is how you add capacity to handle a growing command volume — start more handlers in the same group — without changing anything on the caller's side.

Every subscriber passes the same group: alongside the channel to subscribe_to_commands; the broker tracks membership and picks one live member per command. send_command on the caller side is unaware groups exist — it just blocks for a %CommandReply{}, which comes back from whichever handler happened to process it.

Gotchas: group membership is scoped per channel — subscribers on the same channel with different group names each get their own full copy of every command (fan-out), which looks like a bug when you expected load-balancing. A slow handler still holds up the caller's timeout, since only one worker is ever picked. And if every member of the group is offline when a command arrives, the send simply fails or times out — commands aren't queued or replayed for a group that has no active listener.

Prerequisites

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

Code

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

{:ok, sub1} =
  KubeMQ.Client.subscribe_to_commands(client, channel,
    group: group,
    on_command: fn cmd ->
      IO.puts("[Worker-1] Handling: #{cmd.body}")
      KubeMQ.CommandReply.new(
        request_id: cmd.id, response_to: cmd.reply_channel, executed: true)
    end
  )

{:ok, sub2} =
  KubeMQ.Client.subscribe_to_commands(client, channel,
    group: group,
    on_command: fn cmd ->
      IO.puts("[Worker-2] Handling: #{cmd.body}")
      KubeMQ.CommandReply.new(
        request_id: cmd.id, response_to: cmd.reply_channel, executed: true)
    end
  )

IO.puts("Two command workers in group '#{group}' ready")
Process.sleep(500)

for i <- 1..4 do
  cmd = KubeMQ.Command.new(channel: channel, body: "Task #{i}", timeout: 10_000)

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

KubeMQ.Subscription.cancel(sub1)
KubeMQ.Subscription.cancel(sub2)
KubeMQ.Client.close(client)

How It Works

• The group option distributes commands across handlers in the same consumer group
• Each command is routed to exactly one handler for execution
• Both handlers must return a %CommandReply{} to complete the RPC cycle

Related

• Send Command
• Query Group