Start at Time Delta
Subscribe to a KubeMQ Events Store channel from a relative time offset using the Python SDK.
Overview
A time-delta subscription starts replay from a relative offset — "the last 30 seconds" — instead of a fixed timestamp or sequence number. It's the right tool when a consumer knows how long it was offline but not the exact moment it disconnected: a worker restarting after a deploy, a dashboard reconnecting after a blip, or a batch job that only cares about "recent" history. Computing an absolute cutoff yourself is bookkeeping the broker can do for you.
EventStoreStartPosition.StartAtTimeDelta with events_store_time_delta_seconds passes the offset to the broker, which resolves it to now - delta at subscription time, replays every stored event from that point forward, then hands off to live delivery — the same replay-to-live transition as an absolute-time or sequence-based start.
Gotchas: the delta is evaluated once, server-side, at subscription creation — it does not "slide" as time passes. A delta of zero replays nothing and behaves like starting from new events only. And since the window is wall-clock based, clock skew between producers and the broker can shift which events land inside or outside the boundary.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Start at time delta — subscribe starting from a relative time offset."""
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-at-time-delta-client",
) as client:
# Pre-populate with some messages
for i in range(5):
await client.send_event_store(
EventStoreMessage(
channel="python-events-store.start-at-time-delta",
body=f"Message-{i + 1}".encode(),
)
)
print("Sent 5 messages")
await asyncio.sleep(1)
# Subscribe starting from 30 seconds ago — will replay recent messages
token = AsyncCancellationToken()
async def subscriber() -> None:
async for event in client.subscribe_to_events_store(
subscription=EventsStoreSubscription(
channel="python-events-store.start-at-time-delta",
on_receive_event_callback=lambda e: None,
on_error_callback=lambda e: print(f"Error: {e}"),
events_store_type=EventStoreStartPosition.StartAtTimeDelta,
events_store_time_delta_seconds=30,
),
cancellation_token=token,
):
print(f"Received: {event.body.decode('utf-8')}")
task = asyncio.create_task(subscriber())
await asyncio.sleep(3)
token.cancel()
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
if __name__ == "__main__":
asyncio.run(main())
How It Works
EventStoreStartPosition.StartAtTimeDeltawithevents_store_time_delta_seconds=30replays all events stored in the last 30 seconds relative to when the subscription is created.- The five pre-published messages were stored within that window, so they are all replayed immediately when the subscriber connects.
- Time-delta subscriptions are useful for catch-up scenarios where you want recent history without tracking an explicit sequence or timestamp.
- After replaying the historical window, the subscription continues receiving any new events published to the channel.
Related
Was this page helpful?