Stream Receive

Overview

A downstream receiver is the persistent-connection way to pull queue messages: instead of opening and tearing down a request for every batch, you open one gRPC stream and reuse it across many receive cycles. That matters for any consumer that runs continuously — a worker loop, a background processor — where reconnecting per batch would add latency and churn on both the client and the broker.

receive_queue_messages fetches a batch under manual settlement — no auto_ack — so nothing is removed from the queue until you explicitly settle it. Each message is settled on its own: async_ack() permanently removes it, async_nack() returns it for redelivery, and leaving it unsettled returns it once the visibility timeout expires.

Gotchas: a crash between receiving and settling redelivers the whole batch, so processing must be idempotent; forgetting to settle a message just delays its redelivery until the timeout, it doesn't drop it; and mixing ack/nack per message, as shown here, only makes sense for per-item failures — a batch-wide failure is usually better handled by nacking everything and retrying later.

Prerequisites

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

Code

"""Example: Stream receive — receive messages with ack/reject/requeue options."""

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-stream-receive-client",
    ) as client:
        # Send some test messages
        for i in range(3):
            await client.send_queue_message(
                QueueMessage(
                    channel="python-queues-stream.stream-receive",
                    body=f"Message #{i + 1}".encode(),
                )
            )
        print("Sent 3 test messages")

        # Receive messages via streaming
        response = await client.receive_queue_messages(
            channel="python-queues-stream.stream-receive",
            max_messages=10,
            wait_timeout_seconds=5,
        )

        for msg in response.messages:
            body = msg.body.decode("utf-8")
            print(f"Received: {body}")

            # Demonstrate different acknowledgment strategies
            if "1" in body:
                await msg.async_ack()
                print(f"  -> Acknowledged: {body}")
            elif "2" in body:
                await msg.async_nack()
                print(f"  -> Rejected: {body}")
            else:
                await msg.async_ack()
                print(f"  -> Acknowledged: {body}")


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

How It Works

• receive_queue_messages polls all available messages in a single call (up to max_messages), returning them as a list without auto_ack.
• Each message is settled individually: async_ack() permanently removes it from the queue; async_nack() returns it for redelivery.
• The example demonstrates inline decision logic inside the settlement loop, which is the typical pattern for content-based routing to ack or nack.
• Without auto_ack, the broker holds messages in a visibility-timeout state until settled or until the timeout expires and returns them.

Related

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