Persistent Pub/Sub

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

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

Code

// 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

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

• Pattern overview
• Go SDK Reference
• Cancel Subscription
• Consumer Group