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 response returned by a manual poll: after receiving messages, call response.requeue_all(channel: target_channel) to move every message in that batch 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 runs.

Prerequisites

• KubeMQ server running on localhost:50000
• Ruby SDK installed (gem install kubemq)

Code

require 'kubemq'

address = ENV.fetch('KUBEMQ_ADDRESS', 'localhost:50000')
source_channel = 'ruby-queues.requeue-all'
target_channel = 'ruby-queues.requeue-all-target'

begin
  client = KubeMQ::QueuesClient.new(address: address, client_id: 'qstream-requeue-example')
  puts "Connected to #{address}"

  3.times do |i|
    msg = KubeMQ::Queues::QueueMessage.new(channel: source_channel, metadata: "item-#{i}", body: "data-#{i}")
    client.send_queue_message(msg)
  end
  puts "Pre-sent 3 messages to #{source_channel}"

  receiver = client.create_downstream_receiver
  request = KubeMQ::Queues::QueuePollRequest.new(channel: source_channel, max_items: 5, wait_timeout: 5)
  response = receiver.poll(request)

  puts "Polled #{response.messages.size} messages — requeuing all to #{target_channel}"
  response.requeue_all(channel: target_channel)
  puts 'All messages requeued'

  target_request = KubeMQ::Queues::QueuePollRequest.new(channel: target_channel, max_items: 5, wait_timeout: 5)
  target_response = receiver.poll(target_request)
  puts "Target channel has #{target_response.messages.size} messages"
  target_response.ack_all
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  receiver&.close
  client&.close
  puts 'Done'
end

How It Works

• requeue_all(channel:) moves all messages from the current batch to a different queue channel.
• Messages are removed from the source and placed in the target channel.
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Pattern overview
• Ruby SDK Reference
• Nack All
• Dead Letter Queue