Start from Last

Overview

A subscriber that just restarted usually doesn't need the entire event history — it needs to know where things stand right now without paying the cost of replaying everything that happened while it was offline. EventStoreStartPosition.StartFromLast solves that: it re-anchors a new subscription to the tail of the store, delivering exactly one historical event (the most recently stored one) before switching to live delivery. That's the sweet spot between StartFromNew (no history at all, so you might miss the current state entirely) and StartFromFirst (the full backlog, which can be slow and mostly irrelevant for a consumer that only cares about "now").

Under the hood, EventStoreStartPosition.StartFromLast is passed as the events_store_type on an EventsStoreSubscription. The broker looks up the channel's most recent stored event at subscription time, replays that single event to the new subscriber, and then streams every subsequently published event as it arrives — the same live path any other subscription uses.

Gotchas: if the channel is empty when you subscribe, there's no "last" event to deliver — you simply start receiving new events as they're published, with no error raised. StartFromLast gives you one event, not the last N — if you need a short window of recent history, replay from a sequence number instead. And because "last" is resolved at subscribe time, two subscribers starting a few events apart can each get a different one.

Prerequisites

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

Code

"""Example: StartFromLast — subscribe starting from the last stored event."""

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-from-last-client",
    ) as client:
        # Pre-populate with some messages
        for i in range(5):
            await client.send_event_store(
                EventStoreMessage(
                    channel="python-events-store.start-from-last",
                    body=f"Message-{i + 1}".encode(),
                )
            )
        print("Sent 5 messages")
        await asyncio.sleep(1)


        # Subscribe from the last stored message only
        token = AsyncCancellationToken()

        async def subscriber() -> None:
            async for event in client.subscribe_to_events_store(
                subscription=EventsStoreSubscription(
                    channel="python-events-store.start-from-last",
                    on_receive_event_callback=lambda e: None,
                    on_error_callback=lambda e: print(f"Error: {e}"),
                    events_store_type=EventStoreStartPosition.StartFromLast,
                ),
                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.StartFromLast delivers only the single most recent event stored on the channel at the time of subscription, then switches to live delivery.
• The five pre-published messages are in history; only Message-5 (the last one) is replayed because StartFromLast skips all but the final stored event.
• This is useful for stateful consumers that only need to catch up on the most recent state rather than replaying the full history.
• After delivering the last stored event, the subscription continues receiving any new events published to the channel.

Related

• Pattern overview
• Python SDK Reference
• Persistent Pub/Sub
• Cancel Subscription