Start from First
Subscribe and replay all events from the beginning
Overview
A new consumer joining an Events Store channel usually needs more than what happens next — it needs everything that already happened. StartFromFirstEvent solves that by replaying the channel's complete stored history before switching to live delivery, so a service can rebuild its state from scratch instead of starting with a blank slate and hoping nothing important was missed.
Under the hood, the broker walks the store from the oldest retained sequence forward, streaming each event to your callback in order, then hands off to live delivery of new events without a gap. You don't manage offsets or checkpoints yourself — the start position is set once, at subscription time, via kubemq.StartFromFirstEvent().
Gotchas: on a long-lived channel this can mean replaying millions of events before anything new shows up, so it's the wrong choice for a consumer that only cares about "from now on" (use StartNewOnly for that). Retention and expiration policies still apply — events already purged by TTL or max-count limits are gone and won't be replayed, so "full history" only means what the store still has.
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-first
//
// Demonstrates subscribing to event store with StartFromFirstEvent.
// All stored events from the beginning are replayed, then new events
// continue to be delivered.
//
// Channel: go-events-store.start-from-first
// Client ID: go-events-store-start-from-first-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-first-client"),
)
if err != nil {
log.Fatal(err)
}
defer client.Close()
channel := "go-events-store.start-from-first"
// First, send some events so there is data to replay.
for i := 1; i <= 3; i++ {
_, err := client.SendEventStore(ctx, kubemq.NewEventStore().
SetChannel(channel).
SetBody(fmt.Appendf(nil, "stored-msg-%d", i)).
SetMetadata("replay-test"))
if err != nil {
log.Fatal(err)
}
}
fmt.Println("Sent 3 events to store")
received := make(chan struct{}, 3)
// Subscribe with StartFromFirstEvent — replays all stored events.
sub, err := client.SubscribeToEventsStore(ctx, channel, "",
kubemq.StartFromFirstEvent(),
kubemq.WithOnEventStoreReceive(func(e *kubemq.EventStoreReceive) {
fmt.Printf("[StartFromFirst] 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()
// Wait for events to arrive.
time.Sleep(3 * time.Second)
fmt.Printf("Received %d events from replay\n", len(received))
}
How It Works
- Three events are published before the subscription is created to ensure there is history to replay.
kubemq.StartFromFirstEvent()tells the broker to send every stored event on the channel from the oldest retained sequence forward, then continue with new events.- The subscription streams all replayed events to the
WithOnEventStoreReceivecallback in sequence order. len(received)reads the buffered channel's current length after a short wait to count deliveries without a race condition — a pattern useful in one-shot demos or tests.
Related
Was this page helpful?