Stream Send

Overview

Publishing events one at a time over a fresh call each time caps throughput when you need to push hundreds or thousands of events per second — log forwarding, sensor telemetry, change-data-capture feeds — where per-call overhead dominates. AsyncPubSubClient avoids that cost by keeping events on a persistent, non-blocking path under the hood: publish_event writes onto an internal gRPC stream shared by the client rather than opening a new call per event, so a tight loop of await client.publish_event(...) sends fast without waiting on a broker round-trip for each one.

That streaming path is what lets the loop in this sample publish 100 events back-to-back with no artificial delay between sends — each publish_event call just enqueues the next event on the open connection.

Gotchas: publishing is fire-and-forget — publish_event returning doesn't mean a subscriber received the event, only that it left the client; a subscriber that starts listening late will simply miss events sent before it subscribed. Because sends aren't acknowledged individually, give the broker a moment to forward buffered events to subscribers before you tear down the connection, and watch each subscription's on_error_callback rather than expecting publish_event itself to raise on delivery problems.

Prerequisites

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

Code

"""Example: Stream send — send many events efficiently via the streaming path."""

from __future__ import annotations

import asyncio

from kubemq import (
    AsyncCancellationToken,
    AsyncPubSubClient,
    EventMessage,
    EventsSubscription,
)


async def main() -> None:
    async with AsyncPubSubClient(
        address="localhost:50000",
        client_id="python-events-stream-send-client",
    ) as client:
        token = AsyncCancellationToken()
        received: list[str] = []

        async def subscriber() -> None:
            async for event in client.subscribe_to_events(
                subscription=EventsSubscription(
                    channel="python-events.stream-send",
                    on_receive_event_callback=lambda e: None,
                    on_error_callback=lambda e: print(f"Error: {e}"),
                ),
                cancellation_token=token,
            ):
                received.append(event.body.decode("utf-8"))
                print(f"Received: {event.body.decode('utf-8')}")

        task = asyncio.create_task(subscriber())
        await asyncio.sleep(1)

        for i in range(100):
            await client.publish_event(
                EventMessage(
                    channel="python-events.stream-send",
                    body=f"Event-{i + 1}".encode(),
                )
            )

        print("Sent 100 events via stream")
        await asyncio.sleep(3)
        print(f"Received {len(received)} events")
        token.cancel()
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass


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

How It Works

• publish_event is called in a tight loop with await, sending 100 events sequentially over the same gRPC connection.
• The subscriber coroutine accumulates received events in received; the final count confirms how many were delivered.
• Events are fire-and-forget from the publisher's perspective: publish_event does not wait for subscriber acknowledgement.
• The 3-second sleep after sending gives the broker time to forward all buffered events to the subscriber before the token is cancelled.

Related

• Pattern overview
• Python SDK Reference
• Basic Pub/Sub
• Cancel Subscription