Replay from Time

Overview

Replaying from a timestamp lets a consumer recover a window of history without knowing exact sequence numbers — you reach for it after a deploy, an outage, or any gap where you know roughly when you went dark but not where you left off in the stream. It turns an Events Store channel into a rewindable log: resubscribe with a point in time and the broker replays every event stored at or after it, then hands off to live delivery.

The subscription's start position is set to StartFromTime with a time.Time value — the broker compares this against the storage timestamp it assigned to each event, not any timestamp embedded in the payload. Because it's wall-clock based, the window is approximate rather than exact: pass a time far enough back to be safe.

Gotchas: clock skew between your subscriber's clock and the server's matters — favor a generous buffer over a precise cutoff. Storage timestamps reflect when the broker persisted the event, not when the producer created it, so under load the two can drift. And unlike sequence-based replay, a time-based start position has no way to guarantee "no gaps, no duplicates" across a network hiccup — use StartFromSequence instead if you need exact resumption.

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-time
//
// Demonstrates subscribing to event store with StartFromTime.
// Events are replayed starting from a specific point in time.
//
// Channel: go-events-store.replay-from-time
// Client ID: go-events-store-replay-from-time-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-time-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-events-store.replay-from-time"
	received := make(chan struct{}, 5)

	// Subscribe starting from 1 hour ago — replays events stored in the last hour.
	since := time.Now().Add(-1 * time.Hour)
	sub, err := client.SubscribeToEventsStore(ctx, channel, "",
		kubemq.StartFromTime(since),
		kubemq.WithOnEventStoreReceive(func(e *kubemq.EventStoreReceive) {
			fmt.Printf("[StartFromTime] 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()

	// Send a new event that should be received.
	_, err = client.SendEventStore(ctx, kubemq.NewEventStore().
		SetChannel(channel).
		SetBody([]byte("event within time window")))
	if err != nil {
		log.Fatal(err)
	}

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

How It Works

1. kubemq.StartFromTime(since) takes a time.Time value and instructs the broker to replay all stored events with a broker-assigned timestamp at or after since.
2. Setting since = time.Now().Add(-1 * time.Hour) effectively requests events from the last hour, then transitions to live delivery once caught up.
3. A new event is sent after subscribing to guarantee at least one delivery within the time window regardless of what was stored before.
4. This is useful for event-sourced systems performing catch-up reads after a consumer downtime of known duration.

Related

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