Replay from Time
Replay KubeMQ Events Store messages from a specific timestamp using the Python SDK subscription API.
Overview
Replaying from a timestamp lets a consumer recover a window of history without knowing exact sequence numbers — you reach for it after a deploy, an outage, or any gap where you know roughly when you went dark but not where you left off in the stream. It turns an Events Store channel into a rewindable log: resubscribe with a point in time and the broker replays every event stored at or after it, then hands off to live delivery.
The subscription's events_store_type is set to EventStoreStartPosition.StartAtTime with events_store_start_time given a datetime value — the broker compares this against the storage timestamp it assigned to each event, not any timestamp embedded in the payload. Because it's wall-clock based, the window is approximate rather than exact: pass a time far enough back to be safe.
Gotchas: clock skew between your subscriber's clock and the server's matters — favor a generous buffer over a precise cutoff. Storage timestamps reflect when the broker persisted the event, not when the producer created it, so under load the two can drift. And unlike sequence-based replay, a time-based start position has no way to guarantee "no gaps, no duplicates" across a network hiccup — use EventStoreStartPosition.StartAtSequence instead if you need exact resumption.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Replay from time — subscribe starting from a specific timestamp."""
from __future__ import annotations
import asyncio
from datetime import datetime
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-replay-from-time-client",
) as client:
# Pre-populate with some messages
for i in range(5):
await client.send_event_store(
EventStoreMessage(
channel="python-events-store.replay-from-time",
body=f"Message-{i + 1}".encode(),
)
)
print("Sent 5 messages")
await asyncio.sleep(1)
# Subscribe starting from the current time — will only get new messages
start_time = datetime.now()
token = AsyncCancellationToken()
async def subscriber() -> None:
async for event in client.subscribe_to_events_store(
subscription=EventsStoreSubscription(
channel="python-events-store.replay-from-time",
on_receive_event_callback=lambda e: None,
on_error_callback=lambda e: print(f"Error: {e}"),
events_store_type=EventStoreStartPosition.StartAtTime,
events_store_start_time=start_time,
),
cancellation_token=token,
):
print(f"Received: {event.body.decode('utf-8')}")
task = asyncio.create_task(subscriber())
await asyncio.sleep(1)
# Send a new message after the start time
await client.send_event_store(
EventStoreMessage(
channel="python-events-store.replay-from-time",
body=b"Message after start time",
)
)
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.StartAtTimewithevents_store_start_time=start_timetells the broker to replay all events stored at or after that timestamp.- The five pre-published messages are stored before
start_time, so they fall outside the replay window and are not received. - One message sent after
start_timefalls within the window and is delivered to the subscriber. - After replaying historical events from the given timestamp, the subscription continues receiving any future events published to the channel.
Related
Was this page helpful?