Start New Only

Overview

Start-from-new turns a durable Events Store channel into a live-only feed — reach for it when a consumer only cares what happens from this moment forward and would rather skip a large backlog than pay to replay it. Dashboards, live notification fan-outs, and freshly-deployed services that don't need to catch up on history are the classic cases: any of the replay-from-start positions would mean churning through every historical event just to reach the live tail.

It works by passing StartFromNewEvents() to SubscribeToEventsStore — the broker stamps the subscription's registration time as a watermark and delivers only events published after it, ignoring everything already stored. Gotchas: there's a race between registering and the publisher sending — a publish that lands before the broker fully registers you is silently skipped, so give the subscription a moment to settle before publishing; this position can never see anything published earlier, so use StartFromFirstEvent or StartFromSequence when you need guaranteed replay; and reconnecting doesn't resume where you left off — a fresh StartFromNewEvents() subscription starts from "now" again, with no cursor persisted across restarts.

Prerequisites

• KubeMQ server running on localhost:50000
• Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

// Example: events-store/start-new-only
//
// Demonstrates subscribing to event store with StartFromNewEvents.
// Only events published after the subscription is established are delivered.
// Previously stored events are not replayed.
//
// Channel: go-events-store.start-new-only
// Client ID: go-events-store-start-new-only-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-new-only-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-events-store.start-new-only"
	received := make(chan struct{})

	// Subscribe with StartFromNewEvents — only new events are delivered.
	sub, err := client.SubscribeToEventsStore(ctx, channel, "",
		kubemq.StartFromNewEvents(),
		kubemq.WithOnEventStoreReceive(func(e *kubemq.EventStoreReceive) {
			fmt.Printf("[StartFromNew] seq=%d body=%s\n", e.Sequence, string(e.Body))
			close(received)
		}),
		kubemq.WithOnError(func(err error) {
			log.Println("Subscription error:", err)
		}),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Unsubscribe()

	// Allow time for subscription to register on server
	time.Sleep(1 * time.Second)

	// Send an event after subscribing.
	_, err = client.SendEventStore(ctx, kubemq.NewEventStore().
		SetChannel(channel).
		SetBody([]byte("new event only")).
		SetMetadata("new-only"))
	if err != nil {
		log.Fatal(err)
	}

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

How It Works

1. kubemq.StartFromNewEvents() tells the broker to ignore all pre-existing stored events and deliver only messages published after the subscription is registered.
2. The one-second sleep gives the server time to register the subscription before the publisher sends — without it, the published event might race ahead and miss the subscription window.
3. Even though the events are stored (persistent), this start position behaves like a live events subscription: no history, just the live stream.
4. Use StartFromFirstEvent or StartFromSequence when you need guaranteed replay from a known point in history.

Related

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