Start from First

Overview

A new consumer joining an Events Store channel usually needs more than what happens next — it needs everything that already happened. StartPosition.StartFromFirst solves that by replaying the channel's complete stored history before switching to live delivery, so a service can rebuild its state from scratch instead of starting with a blank slate and hoping nothing important was missed.

Under the hood, the broker walks the store from the oldest retained sequence forward, streaming each event through the subscribeToEventsStore flow in order, then hands off to live delivery of new events without a gap. You don't manage offsets or checkpoints yourself — the start position is set once, at subscription time, via startPosition = StartPosition.StartFromFirst.

Gotchas: on a long-lived channel this can mean replaying millions of events before anything new shows up, so it's the wrong choice for a consumer that only cares about "from now on" (use StartNewOnly for that). Retention and expiration policies still apply — events already purged by TTL or max-count limits are gone and won't be replayed, so "full history" only means what the store still has.

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-start-from-first"
private const val CHANNEL = "kotlin-events-store.start-from-first"

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

    client.use {
        // Publish events
        repeat(5) { i ->
            client.publishEventStore(eventStoreMessage {
                channel = CHANNEL
                body = "History event #${i + 1}".toByteArray()
            })
            println("Published #${i + 1}")
        }

        delay(500)

        // Subscribe from the very first event -- replays complete history
        println("\nSubscribing with StartFromFirst:")
        val subJob = launch {
            client.subscribeToEventsStore {
                channel = CHANNEL
                startPosition = StartPosition.StartFromFirst
            }.take(5).collect { msg ->
                println("  seq=${msg.sequence}: ${String(msg.body)}")
            }
        }

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

How It Works

• StartPosition.StartFromFirst replays all events from sequence 1 onward.
• This provides a complete history replay of every event ever published to the channel.
• New events published after subscription are also delivered.
• Use with caution on channels with large event histories -- consider StartAtSequence or StartAtTimeDelta for targeted replay.

Related

• Start from Last
• Start New Only
• Events Store Reference