Requeue All

Overview

Requeue all moves an entire batch of polled messages to a different channel in one server-side operation, without republishing them from the client. Reach for it when you need to make a routing decision after looking at a batch — shovel a stuck batch into a review queue, redirect it to a priority pipeline, or migrate messages off a channel that's being retired, all while the source queue is cleared atomically.

It works against the poll result returned by poll_queue: after receiving messages, call KubeMQ.PollResponse.requeue_all(poll, target_channel) to move every message in that result to the target channel in one call, removing them from the source at the same instant. The messages keep their original body, tags, and policies — the broker relocates them, it doesn't recreate them.

Gotchas: requeuing is all-or-nothing for the batch — there's no per-message filter, so split the batch yourself first if only some messages should move. The destination channel is an ordinary queue with no special semantics; nothing consumes it automatically. And the operation only affects messages still held from that poll — anything already acked or expired beforehand is gone before requeue_all/2 runs.

Prerequisites

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

Code

source_channel = "elixir-queues.requeue-all.source"
target_channel = "elixir-queues.requeue-all.target"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-qs-requeue")

for i <- 1..3 do
  {:ok, _} = KubeMQ.Client.send_queue_message(client,
    KubeMQ.QueueMessage.new(channel: source_channel, body: "Requeue msg #{i}"))
end

IO.puts("Sent 3 messages to '#{source_channel}'")

case KubeMQ.Client.poll_queue(client,
       channel: source_channel,
       max_items: 10,
       wait_timeout: 5_000
     ) do
  {:ok, poll} ->
    IO.puts("Polled #{length(poll.messages)} messages from source")

    case KubeMQ.PollResponse.requeue_all(poll, target_channel) do
      {:ok, _} -> IO.puts("All messages requeued to '#{target_channel}'")
      {:error, err} -> IO.puts("Requeue failed: #{err.message}")
    end

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

case KubeMQ.Client.receive_queue_messages(client, target_channel,
       max_messages: 10,
       wait_timeout: 3_000
     ) do
  {:ok, result} ->
    IO.puts("Target queue has #{result.messages_received} messages")

  {:error, _} ->
    IO.puts("No messages in target")
end

KubeMQ.Client.close(client)

How It Works

• PollResponse.requeue_all/2 moves all polled messages to a different queue channel
• Messages are removed from the source queue and appear in the target queue
• Useful for manual dead-letter routing or priority queue management

Related

• Nack All
• Dead Letter Queue