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. StartFromLastEvent 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, kubemq.StartFromLastEvent() is passed as a subscription option to SubscribeToEventsStore. 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. StartFromLastEvent 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
• Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

// Example: events-store/start-from-last
//
// Demonstrates subscribing to event store with StartFromLastEvent.
// The last stored event is replayed, then new events continue.
//
// Channel: go-events-store.start-from-last
// Client ID: go-events-store-start-from-last-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/kubemq-io/kubemq-go/v2"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	client, err := kubemq.NewClient(ctx,
		kubemq.WithAddress("localhost", 50000),
		kubemq.WithClientId("go-events-store-start-from-last-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-events-store.start-from-last"

	// Send some events so there is a "last" event.
	for i := 1; i <= 3; i++ {
		_, err := client.SendEventStore(ctx, kubemq.NewEventStore().
			SetChannel(channel).
			SetBody(fmt.Appendf(nil, "msg-%d", i)))
		if err != nil {
			log.Fatal(err)
		}
	}
	fmt.Println("Sent 3 events")

	received := make(chan struct{})

	// Subscribe with StartFromLastEvent — starts from the most recent stored event.
	sub, err := client.SubscribeToEventsStore(ctx, channel, "",
		kubemq.StartFromLastEvent(),
		kubemq.WithOnEventStoreReceive(func(e *kubemq.EventStoreReceive) {
			fmt.Printf("[StartFromLast] seq=%d body=%s\n", e.Sequence, string(e.Body))
			select {
			case received <- struct{}{}:
			default:
			}
		}),
		kubemq.WithOnError(func(err error) {
			log.Println("Subscription error:", err)
		}),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Unsubscribe()

	select {
	case <-received:
		fmt.Println("Start from last demo complete")
	case <-ctx.Done():
		log.Fatal("Timed out")
	}
}

How It Works

1. Three events are sent first so the broker has a non-empty store with a known "last" event (sequence 3).
2. kubemq.StartFromLastEvent() instructs the broker to deliver the most recent stored event (sequence 3) and then continue with any new events going forward.
3. Unlike StartFromFirstEvent, older events are not replayed — useful when a consumer just needs to re-anchor to the current tail of the store after a restart.
4. The unbuffered received channel and the select statement ensure the demo waits for at least one delivery before exiting.

Related

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