Overview

Which to use

This page covers high-throughput Events Store (persistent, replayable) streaming via send_event_store. For the fire-and-forget equivalent, see Events Stream Send.

Stream send covers publishing a batch of persistent events back-to-back, without pausing your producer between messages. The single-shot send_event_store call is fine for one-off writes, but if you're bulk-loading history, replicating a firehose of records, or backfilling an Events Store channel, sending events in rapid succession — while a subscriber consumes them concurrently — turns network latency into your throughput ceiling instead of an app-level bottleneck.

Each send_event_store call still confirms storage synchronously, returning an EventStoreResult with the broker-assigned id and a sent flag, and each persisted event gets a monotonically increasing sequence number. Running the sends in a loop while a concurrent subscribe_to_events_store task drains the channel pipelines production and consumption instead of serializing them. Gotchas: each call still blocks on its own confirmation, so very high fan-out workloads benefit from concurrent sends rather than one tight loop; the subscriber must be running before you publish if you want every event, since Events Store subscriptions don't retroactively grab messages sent before they connected; and don't rely on wall-clock sleeps to guarantee delivery — use the returned sequence instead.

Prerequisites

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

Code

stream_send.py

"""Example: Stream send — send events store messages via the bidirectional stream path."""

from __future__ import annotations

import asyncio

from kubemq import (
    AsyncCancellationToken,
    AsyncPubSubClient,
    EventStoreMessage,
    EventsStoreSubscription,
)
from kubemq.pubsub import EventStoreStartPosition


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

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

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

        for i in range(10):
            result = await client.send_event_store(
                EventStoreMessage(
                    channel="python-events-store.stream-send",
                    body=f"StoreEvent-{i + 1}".encode(),
                )
            )
            print(f"Sent: id={result.id}, sent={result.sent}")

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


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

How It Works

Each send_event_store call returns an EventStoreResult with id and sent=True, confirming persistence on the broker.
The subscriber accumulates received events in received; the final count verifies delivery of all 10 persisted messages.
Messages are sent sequentially in the loop; the broker persists each one atomically and assigns an increasing sequence number.
The 3-second sleep after sending allows the broker to forward all queued events to the subscriber before the token is cancelled.

Stream Send

Overview

Prerequisites

Code

How It Works

Related

On this page