Batch Send

Overview

Batch send groups several queue messages into one call instead of sending them one at a time. Reach for it when publishing many related items together — importing records, fanning out a set of jobs, replaying a backlog — since sending each message individually pays a full round trip per message, while batching amortizes that cost across the whole set.

It works by building a list of %QueueMessage{} structs, then passing the list to KubeMQ.Client.send_queue_messages/2 in a single call. The broker enqueues each message independently and returns a %QueueBatchResult{} with a results list — one entry per message, each carrying message_id and is_error — plus a top-level have_errors flag.

Gotchas: have_errors only tells you whether something failed, not what — walk results and check is_error on each entry to find the specific message; a batch is still one bounded request, so it doesn't help continuous, open-ended publishing (use a stream-based send for that); and very large batches raise the size and latency of that single call, so there's a practical ceiling before splitting into multiple batches pays off.

Prerequisites

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

Code

channel = "elixir-queues.batch-send"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-queue-batch")

messages =
  for i <- 1..5 do
    KubeMQ.QueueMessage.new(
      channel: channel,
      body: "Batch message #{i}",
      tags: %{"index" => "#{i}"}
    )
  end

IO.puts("Sending batch of #{length(messages)} messages...")

case KubeMQ.Client.send_queue_messages(client, messages) do
  {:ok, result} ->
    IO.puts("Batch ID: #{result.batch_id}")
    IO.puts("Has errors: #{result.have_errors}")

    Enum.each(result.results, fn r ->
      IO.puts("  Message #{r.message_id}: error=#{r.is_error}")
    end)

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

case KubeMQ.Client.receive_queue_messages(client, channel,
       max_messages: 10,
       wait_timeout: 5_000
     ) do
  {:ok, result} ->
    IO.puts("Received #{result.messages_received} messages from batch")

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

KubeMQ.Client.close(client)

How It Works

• send_queue_messages/2 accepts a list of %QueueMessage{} structs
• Returns %QueueBatchResult{} with individual results for each message
• have_errors indicates whether any messages in the batch failed
• Tags can be used to attach metadata to individual messages

Related

• Send & Receive
• Stream Send