Persistent Pub/Sub
Publish and subscribe to a KubeMQ Events Store channel with message persistence using the Python SDK.
Overview
This tutorial builds a publisher and subscriber on a KubeMQ Events Store channel — reach for this pattern when a subscriber can't guarantee it's listening the instant a message is published. Plain events are fire-and-forget: publish with no one subscribed and the message is gone. Events Store persists every event to a durable, ordered log, so a subscriber connecting seconds or a full restart later still catches up — useful for anything needing a complete history, like an audit trail or event-sourced state.
The two calls involved: send_event_store publishes and returns an EventStoreResult confirming storage plus a broker-assigned sequence number, and subscribe_to_events_store takes a required start position telling the broker where to start — new events only (EventStoreStartPosition.StartFromNew, used here), from the first stored event, or a given sequence or time. Production subscribers usually resume from a saved checkpoint instead of starting fresh.
Gotchas: starting from new events means anything published earlier is silently skipped — this sample papers over that race with a fixed asyncio.sleep instead of a ready signal, fine for a demo but not production. Replaying from the first event on every restart replays the whole log, which gets costly on a busy channel. Persistence isn't consumer coordination: each independent subscriber gets its own full replay unless grouped with a consumer group.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Persistent pub/sub — publish and subscribe to events store with persistence."""
from __future__ import annotations
import asyncio
from kubemq import (
AsyncCancellationToken,
AsyncPubSubClient,
EventStoreMessage,
EventStoreReceived,
EventsStoreSubscription,
KubeMQConnectionError,
KubeMQError,
)
from kubemq.pubsub import EventStoreStartPosition
async def main() -> None:
try:
async with AsyncPubSubClient(
address="localhost:50000",
client_id="python-events-store-persistent-pubsub-client",
) as client:
token = AsyncCancellationToken()
async def subscriber() -> None:
async for event in client.subscribe_to_events_store(
subscription=EventsStoreSubscription(
channel="python-events-store.persistent-pubsub",
on_receive_event_callback=lambda e: None,
on_error_callback=lambda e: print(f"Error: {e}"),
events_store_type=EventStoreStartPosition.StartFromNew,
),
cancellation_token=token,
):
print(
f"Received — Id:{event.id}, Seq:{event.sequence}, "
f"Body:{event.body.decode('utf-8')}"
)
task = asyncio.create_task(subscriber())
await asyncio.sleep(1)
result = await client.send_event_store(
EventStoreMessage(
channel="python-events-store.persistent-pubsub",
body=b"hello kubemq",
)
)
print(f"Send result: {result}")
await asyncio.sleep(2)
token.cancel()
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
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:
# Received — Id:<message-id>, Seq:<sequence>, Body:hello kubemq
# Send result: <result>
How It Works
send_event_storereturns anEventStoreResultwith anidandsentflag, confirming the event was persisted on the broker.EventStoreStartPosition.StartFromNewmeans the subscriber only receives events published after the subscription is established.- The
subscribe_to_events_storeasync generator yieldsEventStoreReceivedobjects that include a monotonically increasingsequencenumber. - Unlike plain events, stored events survive subscriber restarts; a new subscriber can replay history by changing the start position.
Related
Was this page helpful?