Stream Send

Overview

Sending one queue message per call works fine for occasional traffic, but each call carries its own round trip. At high volume — event ingestion, sensor telemetry, log shipping — that per-call overhead caps your throughput well below what the connection can support.

AsyncQueuesClient.send_queue_message reuses the client's persistent gRPC connection across calls, so repeated sends avoid reconnecting per message. Each call returns a QueueSendResult carrying the broker-assigned id and an is_error flag, and the message is durably persisted before the call returns — giving per-message durability even in a tight loop. tags let downstream consumers filter or route by attributes like batch or sensor ID without parsing the body.

Gotchas: send_queue_message awaits each send in turn, so throughput is bounded by round-trip latency — for high volumes, pipeline in-flight sends with asyncio.gather instead of a strict sequential loop. Always check result.is_error; a broker-side rejection doesn't raise, it's just a flag, so an ignored error means a message never enqueued. Use the client inside async with — letting it go out of scope without closing leaves the connection open.

Prerequisites

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

Code

"""Example: Stream send — send multiple messages via the streaming interface."""

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-send-client",
    ) as client:
        for i in range(5):
            result = await client.send_queue_message(
                QueueMessage(
                    channel="python-queues-stream.stream-send",
                    body=f"Stream message #{i + 1}".encode(),
                    tags={"batch": "upstream-demo", "index": str(i)},
                )
            )
            print(f"Sent message #{i + 1}: ID={result.id}, Error={result.is_error}")

        print("All messages sent via upstream stream")


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

How It Works

• send_queue_message is called in a loop; each call returns a QueueSendResult with the broker-assigned id and an is_error flag.
• tags are key-value string pairs attached to the message for routing, filtering, and tracing at the broker level.
• Each message is persisted by the broker before the next iteration begins, giving per-message durability guarantees.
• is_error should be checked per result to detect broker rejections without interrupting the rest of the batch.

Related

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