# Persistent Pub/Sub (/sdks/go/tutorials/persistent-pubsub)



## Overview [#overview]

This tutorial builds a publisher and subscriber on a KubeMQ Events Store channel — reach for this pattern when a subscriber can't guarantee it's listening the instant a message is published. Plain events are fire-and-forget: publish with no one subscribed and the message is gone. Events Store persists every event to a durable, ordered log, so a subscriber connecting seconds or a full restart later still catches up — useful for anything needing a complete history, like an audit trail or event-sourced state.

The two calls involved: `SendEventStore` publishes and returns a result confirming storage plus a broker-assigned sequence number, and `SubscribeToEventsStore` takes a required `SubscriptionOption` telling the broker where to start — new events only (`StartFromNewEvents`, used here), from the first stored event, or a given sequence or time. Production subscribers usually resume from a saved checkpoint instead of starting fresh.

**Gotchas:** starting from new events means anything published earlier is silently skipped — this sample papers over that race with a fixed `time.Sleep` instead of a ready signal, fine for a demo but not production. Replaying from the first event on every restart replays the whole log, which gets costly on a busy channel. Persistence isn't consumer coordination: each independent subscriber gets its own full replay unless grouped with a consumer group.

## Prerequisites [#prerequisites]

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

## Code [#code]

```go title="main.go"
// Example: events-store/persistent-pubsub
//
// Demonstrates basic persistent event store publish/subscribe.
// Events are stored and can be replayed. This example sends an event
// and subscribes with StartFromNewEvents to receive only new events.
//
// Channel: go-events-store.persistent-pubsub
// Client ID: go-events-store-persistent-pubsub-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-persistent-pubsub-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-events-store.persistent-pubsub"
	received := make(chan struct{})

	// Subscribe to new events on the store channel.
	sub, err := client.SubscribeToEventsStore(ctx, channel, "",
		kubemq.StartFromNewEvents(),
		kubemq.WithOnEventStoreReceive(func(e *kubemq.EventStoreReceive) {
			fmt.Printf("Received: 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 subscription to fully establish before publishing.
	time.Sleep(time.Second)

	// Send a persistent event.
	result, err := client.SendEventStore(ctx, kubemq.NewEventStore().
		SetChannel(channel).
		SetBody([]byte("persistent hello")).
		SetMetadata("greeting"))
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Event stored: id=%s sent=%v\n", result.Id, result.Sent)

	select {
	case <-received:
		fmt.Println("Persistent pub/sub demo complete")
	case <-ctx.Done():
		log.Fatal("Timed out waiting for event")
	}
}

// Expected output:
// Event stored: id=<message-id> sent=true
// Received: seq=<sequence> body=persistent hello
// Persistent pub/sub demo complete

```

## How It Works [#how-it-works]

1. `SubscribeToEventsStore` requires a `SubscriptionOption` as the fourth argument — here `kubemq.StartFromNewEvents()` — which tells the broker to deliver only events published after the subscription is registered.
2. `WithOnEventStoreReceive` registers the callback; each delivery includes a broker-assigned `Sequence` number and timestamps for ordering and deduplication.
3. `SendEventStore` returns `*EventStoreResult` with the broker-assigned `Id` and the `Sent` flag confirming durable storage.
4. Unlike plain events, stored events survive restarts; replay from different start positions (e.g. `StartFromFirstEvent`, `StartFromSequence`) lets late-joining subscribers catch up.

## Related [#related]

* [Pattern overview](/learn/events-store/getting-started)
* [Go SDK Reference](/sdks/go/reference)
* [Cancel Subscription](/sdks/go/how-to/events-store/cancel-subscription)
* [Consumer Group](/sdks/go/how-to/events-store/consumer-group)