Start New Only
Subscribe to a KubeMQ Events Store channel for new events only, skipping stored history, in Python.
Overview
Start-from-new turns a durable Events Store channel into a live-only feed — reach for it when a consumer only cares what happens from this moment forward and would rather skip a large backlog than pay to replay it. Dashboards, live notification fan-outs, and freshly-deployed services that don't need to catch up on history are the classic cases: any of the replay-from-start positions would mean churning through every historical event just to reach the live tail.
It works by passing EventStoreStartPosition.StartFromNew as the events_store_type on subscribe_to_events_store — the broker stamps the subscription's registration time as a watermark and delivers only events published after it, ignoring everything already stored. Gotchas: there's a race between registering and the publisher sending — a publish that lands before the broker fully registers you is silently skipped, so give the subscription a moment to settle before publishing; this position can never see anything published earlier, so use a start-from-first or start-from-sequence position when you need guaranteed replay; and reconnecting doesn't resume where you left off — a fresh StartFromNew subscription starts from "now" again, with no cursor persisted across restarts.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: StartFromNew — subscribe only to new events published after subscription."""
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-start-new-only-client",
) as client:
# Send messages BEFORE subscribing — these will NOT be received
for i in range(3):
await client.send_event_store(
EventStoreMessage(
channel="python-events-store.start-new-only",
body=f"Old-Message-{i + 1}".encode(),
)
)
print("Sent 3 old messages before subscribing")
token = AsyncCancellationToken()
async def subscriber() -> None:
async for event in client.subscribe_to_events_store(
subscription=EventsStoreSubscription(
channel="python-events-store.start-new-only",
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: {event.body.decode('utf-8')}")
task = asyncio.create_task(subscriber())
await asyncio.sleep(1)
# Send messages AFTER subscribing — only these will be received
await client.send_event_store(
EventStoreMessage(
channel="python-events-store.start-new-only",
body=b"New message after subscription",
)
)
print("Sent 1 new message after subscribing")
await asyncio.sleep(2)
token.cancel()
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
if __name__ == "__main__":
asyncio.run(main())
How It Works
EventStoreStartPosition.StartFromNewskips all previously stored events; the subscriber only receives events published after the subscription is established.- The three "old" messages are already persisted when the subscriber connects, so they fall outside the delivery window and are never received.
- The one "new" message is published after the subscription is active, so it is delivered and printed.
- This is the default mode for fresh consumers that want real-time events without replaying any history.
Related
Was this page helpful?