Start New Only
Receive only new events published after subscription
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 setting startPosition = StartPosition.StartNewOnly in the subscribeToEventsStore builder — 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 StartPosition.StartFromFirst or a sequence-based position when you need guaranteed replay; and reconnecting doesn't resume where you left off — a fresh StartNewOnly subscription starts from "now" again, with no cursor persisted across restarts.
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-new-only"
private const val CHANNEL = "kotlin-events-store.start-new-only"
fun main() = runBlocking {
val client = KubeMQClient.pubSub {
address = ADDRESS
clientId = CLIENT_ID
}
client.use {
// Publish events BEFORE subscription
repeat(3) { i ->
client.publishEventStore(eventStoreMessage {
channel = CHANNEL
body = "Old event #${i + 1}".toByteArray()
})
println("Published old event #${i + 1} (before subscription)")
}
delay(500)
// Subscribe with StartNewOnly -- should NOT receive old events
println("\nSubscribing with StartNewOnly (old events should not appear):")
val subJob = launch {
client.subscribeToEventsStore {
channel = CHANNEL
startPosition = StartPosition.StartNewOnly
}.take(2).collect { msg ->
println(" seq=${msg.sequence}: ${String(msg.body)}")
}
}
delay(500)
// Publish NEW events after subscription
repeat(2) { i ->
client.publishEventStore(eventStoreMessage {
channel = CHANNEL
body = "New event #${i + 1}".toByteArray()
})
println("Published new event #${i + 1} (after subscription)")
}
subJob.join()
println("Done.")
}
}How It Works
StartPosition.StartNewOnlyskips all historical events and only delivers events published after subscription.- Old events (#1-3) are not delivered; only new events (#1-2 published after subscription) appear.
- This behaves like regular events but with persistence -- if the subscriber disconnects and reconnects with
StartFromFirst, it can replay the history.
Related
Was this page helpful?