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 batch returned by a manual poll on the downstream receiver: after polling with auto_ack: false, call poll.re_queue_all(target_channel) to send a batch RequeueAll settlement request that moves every message in that batch to the target channel, 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 in that poll batch — anything already acked or expired beforehand is gone before re_queue_all runs.

Prerequisites

• KubeMQ server running on localhost:50000
• Rust SDK installed (cargo add kubemq)

Code

use kubemq::prelude::*;
use kubemq::{PollRequest, QueueMessageBuilder};

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .build()
        .await?;

    let source_channel = "rust-queues.requeue-all";
    let target_channel = "rust-queues.requeue-all-target";

    let msg = QueueMessageBuilder::new()
        .channel(source_channel)
        .body(b"requeue-me".to_vec())
        .build();
    client.send_queue_message(msg).await?;
    println!("Sent message to source queue");

    let mut receiver = client.new_queue_downstream_receiver().await?;
    let poll = receiver.poll(PollRequest {
        channel: source_channel.to_string(),
        max_items: 10,
        wait_timeout_seconds: 5,
        auto_ack: false,
    }).await?;
    println!("Polled {} messages", poll.messages.len());

    poll.re_queue_all(target_channel).await?;
    println!("Re-queued all messages to '{}'", target_channel);

    receiver.close().await?;
    client.close().await?;
    Ok(())
}

How It Works

• new_queue_downstream_receiver() opens a persistent bidirectional gRPC stream for transactional polling.
• poll() with auto_ack: false fetches messages without consuming them.
• re_queue_all(target) sends a batch RequeueAll settlement request, routing all polled messages to the specified target channel instead of their original channel.
• Review timeouts, channel names, and client IDs before running against shared environments.
• Run the program while the server from the prerequisites is available.

Related

• Queues overview
• Rust SDK Reference
• Nack All