Poll Mode

Overview

Poll mode is a pull-based way to consume queue messages: the consumer decides exactly when to ask for work and how much, instead of holding an open stream the broker pushes into. That control matters for batch jobs, cron-triggered workers, and any consumer that only runs intermittently and would rather ask "is there anything for me?" than keep a subscription alive.

A single call to receive_queue_messages sends a channel, max_messages, and wait_timeout_seconds; the broker holds the request open as a long poll and returns once enough messages are available or the timeout elapses, so the call never spins on an empty queue. auto_ack=True settles the whole batch on delivery, with no separate acknowledgment step.

Gotchas: auto-ack removes messages the instant they're delivered — a crash mid-processing loses them, so pass auto_ack=False and ack manually when work can fail; the timeout bounds latency, not throughput, so a small max_messages on a busy queue means many round trips; and always check response.is_error before touching response.messages — an empty, non-error response just means nothing arrived during the wait window.

Prerequisites

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

Code

"""Example: Poll mode — single-shot polling pattern for queue messages."""

from __future__ import annotations

import asyncio

from kubemq import AsyncQueuesClient
from kubemq import QueueMessage


async def main() -> None:
    async with AsyncQueuesClient(
        address="localhost:50000",
        client_id="python-queues-stream-poll-mode-client",
    ) as client:
        await client.send_queue_message(
            QueueMessage(
                channel="python-queues-stream.poll-mode",
                body=b"message for polling",
            )
        )
        print("Sent 1 message")

        response = await client.receive_queue_messages(
            channel="python-queues-stream.poll-mode",
            max_messages=10,
            wait_timeout_seconds=5,
            auto_ack=True,
        )

        if response.is_error:
            print(f"Error: {response.error}")
            return

        if not response.messages:
            print("No messages available")
            return

        print(f"Polled {len(response.messages)} messages:")
        for msg in response.messages:
            print(f"  id={msg.id}, body={msg.body.decode('utf-8')}")


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

How It Works

• receive_queue_messages with auto_ack=True polls the channel, delivers all available messages (up to max_messages), and acknowledges them automatically.
• The method returns immediately once wait_timeout_seconds elapses with fewer messages than requested, so it always returns within that window.
• response.is_error should be checked before processing; a non-error empty response means no messages were available during the wait window.
• Poll mode is the simplest consumption pattern: call it on a schedule or in a loop, process what arrives, and repeat.

Related

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