Send & Receive

Overview

Queue send/receive is the foundational operation for guaranteed-delivery, point-to-point messaging: you reach for it whenever work needs to survive past the moment it's created and be handled by exactly one consumer, not broadcast to every subscriber. Unlike pub/sub, a queued message sits durably on the broker until something pulls it, so the producer and consumer never need to be online at the same time — a slow or offline worker adds latency, it doesn't drop the message.

This tutorial builds the smallest possible version of that round trip: send_queue_message enqueues a message on a channel, and receive_queue_messages pulls it back within a bounded wait_timeout_seconds. Settlement here is manual — each received message must be confirmed with msg.async_ack() once your handler finishes; nothing is removed from the queue automatically.

Gotchas: if you forget to call async_ack() (or your handler crashes first), the message stays in the queue and becomes available for redelivery again once its visibility timeout elapses — write handlers that tolerate seeing the same message twice. Calling receive_queue_messages against an empty queue isn't an error; it just blocks until wait_timeout_seconds elapses and returns no messages. And a non-empty error field on the QueueSendResult means the broker rejected the send — check it explicitly rather than assuming success.

Prerequisites

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

Code

"""Example: Send and receive — basic queue message send and receive with manual ack."""

from __future__ import annotations

import asyncio

from kubemq import AsyncQueuesClient
from kubemq import KubeMQConnectionError, KubeMQError, QueueMessage


async def main() -> None:
    try:
        async with AsyncQueuesClient(
            address="localhost:50000",  # TODO: Replace with your KubeMQ server address
            client_id="python-queues-send-receive-client",
        ) as client:
            # Send a message
            result = await client.send_queue_message(
                QueueMessage(
                    channel="python-queues.send-receive",
                    body=b"Hello from queue!",
                    metadata="example-metadata",
                    tags={"source": "send_receive_example"},
                )
            )
            print(f"Sent: id={result.id}, sent_at={result.sent_at}, error={result.error}")

            # Receive and acknowledge the message
            response = await client.receive_queue_messages(
                channel="python-queues.send-receive",
                max_messages=1,
                wait_timeout_seconds=10,
            )
            for msg in response.messages:
                print(f"Received: id={msg.id}, body={msg.body.decode('utf-8')}")
                await msg.async_ack()
                print("  Message acknowledged")
    except KubeMQConnectionError as e:
        print(f"Connection error: {e}")
    except KubeMQError as e:
        print(f"KubeMQ error: {e}")


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

# Expected output:
# Sent: id=<message-id>, sent_at=<timestamp>, error=
# Received: id=<message-id>, body=Hello from queue!
#   Message acknowledged

How It Works

• send_queue_message returns a QueueSendResult with id, sent_at, and error; a non-empty error field means the send was rejected by the broker.
• receive_queue_messages blocks until max_messages are available or wait_timeout_seconds elapses, then returns all collected messages.
• Each received message must be explicitly acknowledged via msg.async_ack(); without an ack the message remains in the queue and becomes available again after its visibility timeout.
• AsyncQueuesClient as an async context manager ensures the underlying gRPC connection is cleanly closed on exit.

Related

• Pattern overview
• Python SDK Reference
• Ack All
• Ack & Reject