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.re_queue_all(destination) to move every polled message to the destination channel in one broker 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. Guard the call with response.is_empty(); calling re_queue_all on an empty poll result is a wasted round trip. And the destination channel is an ordinary queue with no special semantics — nothing consumes it automatically.

Prerequisites

• KubeMQ server running on localhost:50000
• Python SDK installed (pip install kubemq)

Code

"""Example: Requeue all — re-queue all received messages to a different channel."""

from __future__ import annotations

import asyncio

from kubemq.queues import AsyncClient as AsyncQueuesClient
from kubemq import QueueMessage


async def main() -> None:
    async with AsyncQueuesClient(
        address="localhost:50000",
        client_id="python-queues-stream-requeue-all-client",
    ) as client:
        await client.send_queue_message(
            QueueMessage(
                channel="python-queues-stream.requeue-source",
                body=b"move-me",
            )
        )

        response = await client.receive_queue_messages(
            channel="python-queues-stream.requeue-source",
            max_messages=10,
            wait_timeout_seconds=5,
        )
        print(f"Received {response.count()} messages from source")
        if not response.is_empty():
            await response.re_queue_all("python-queues-stream.requeue-destination")
            print("All messages re-queued to destination")

        dest = await client.receive_queue_messages(
            channel="python-queues-stream.requeue-destination",
            max_messages=10,
            wait_timeout_seconds=5,
        )
        print(f"Destination has {dest.count()} messages")


if __name__ == "__main__":
    asyncio.run(main())

How It Works

• response.re_queue_all(destination) atomically moves all polled messages to a different channel in a single broker operation.
• response.is_empty() guards against calling re_queue_all when no messages were received; response.count() reports how many were moved.
• Re-queuing does not re-send: the original messages (with their metadata, tags, and policies) are transferred to the destination channel.
• This pattern is used for manual DLQ drains, channel migrations, and routing failures to a retry queue under explicit operator control.

Related

• Pattern overview
• Python SDK Reference
• Send & Receive
• Ack All