Replay from Sequence
Replay events starting from a specific sequence number
Overview
Replaying from a sequence number lets a consumer resume an events-store subscription from an exact point in a channel's history, instead of re-reading everything or only catching new traffic. It's the checkpoint-recovery pattern: a worker persists the last sequence it processed, and after a crash or redeploy it reopens the subscription right there — no gap, no reprocessing everything that came before.
Sequence numbers are broker-assigned per channel, starting at 1 and increasing monotonically with every stored event; they never reset unless the channel is purged. kubemq.StartFromSequence(n) tells the broker to begin delivery at sequence n inclusive, replaying stored events from that point, then transitioning the subscription to live delivery for anything published afterward.
Gotchas: the sequence is inclusive, so StartFromSequence(3) still delivers event 3 — off by one and you'll reprocess or silently drop a message; you must track and persist the "last processed" sequence yourself, KubeMQ doesn't checkpoint it for you; and requesting a sequence past the current head isn't an error — you'll just get nothing until new events catch up to it.
Prerequisites
- KubeMQ server running on
localhost:50000 - Go SDK installed (
go get github.com/kubemq-io/kubemq-go/v2)
Code
// Example: events-store/replay-from-sequence
//
// Demonstrates subscribing to event store with StartFromSequence.
// Events are replayed starting from a specific sequence number.
//
// Channel: go-events-store.replay-from-sequence
// Client ID: go-events-store-replay-from-sequence-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-replay-from-sequence-client"),
)
if err != nil {
log.Fatal(err)
}
defer client.Close()
channel := "go-events-store.replay-from-sequence"
// Send some events to build up sequence numbers.
for i := 1; i <= 5; 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 5 events")
received := make(chan struct{}, 5)
// Subscribe starting at sequence 3 — replays events from seq 3 onward.
sub, err := client.SubscribeToEventsStore(ctx, channel, "",
kubemq.StartFromSequence(3),
kubemq.WithOnEventStoreReceive(func(e *kubemq.EventStoreReceive) {
fmt.Printf("[StartFromSeq(3)] 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()
time.Sleep(3 * time.Second)
fmt.Printf("Received %d events from sequence 3 onward\n", len(received))
}
How It Works
- Five events are published first to establish a known sequence range on the channel.
kubemq.StartFromSequence(3)tells the broker to begin delivery at sequence number 3, skipping the first two stored events.- The broker replays sequences 3, 4, and 5, then continues delivering any new events published afterwards — the subscription is live, not a one-shot read.
len(received)counts delivered events by draining the buffered channel — a simple way to verify replay count without a separate counter variable.
Related
Was this page helpful?