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: publishEventStore() publishes and returns a result confirming storage with a sent flag plus a broker-assigned sequence number, and subscribeToEventsStore { } requires a startPosition telling the broker where to start — new events only (StartPosition.StartNewOnly, 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 delay 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
• Kotlin SDK installed (implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))

Code

package io.kubemq.sdk.examples.eventsstore

import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.pubsub.StartPosition
import io.kubemq.sdk.pubsub.eventStoreMessage
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-events-store-persistent-pubsub"
private const val CHANNEL = "kotlin-events-store.persistent-pubsub"

fun main() = runBlocking {
    val client = KubeMQClient.pubSub {
        address = ADDRESS
        clientId = CLIENT_ID
    }

    client.use {
        // Subscribe with StartNewOnly
        val subJob = launch {
            client.subscribeToEventsStore {
                channel = CHANNEL
                startPosition = StartPosition.StartNewOnly
            }.take(3).collect { msg ->
                println("Received: seq=${msg.sequence} ${String(msg.body)}")
            }
        }

        delay(500)

        // Publish persistent events
        repeat(3) { i ->
            val result = client.publishEventStore(eventStoreMessage {
                channel = CHANNEL
                body = "Persistent event #$i".toByteArray()
                metadata = "persistence-demo"
            })
            println("Published #$i, sent=${result.sent}")
        }

        subJob.join()
        println("Done.")
    }
}

How It Works

• publishEventStore() persists the event to durable storage and returns a result with sent flag.
• subscribeToEventsStore { } returns a Flow with a required startPosition for replay control.
• Each received event includes a sequence number for ordering.
• Events Store messages are durable and survive broker restarts.

Related

• Events Store Reference
• Start from First
• Replay from Sequence