Start from Last
Subscribe to a KubeMQ events store channel from the most recent event plus new ones with StartFromLast in Kotlin.
Overview
A subscriber that just restarted usually doesn't need the entire event history — it needs to know where things stand right now without paying the cost of replaying everything that happened while it was offline. StartPosition.StartFromLast solves that: it re-anchors a new subscription to the tail of the store, delivering exactly one historical event (the most recently stored one) before switching to live delivery. That's the sweet spot between StartFromNew (no history at all, so you might miss the current state entirely) and StartFromFirst (the full backlog, which can be slow and mostly irrelevant for a consumer that only cares about "now").
Under the hood, startPosition = StartPosition.StartFromLast is passed to the events store subscription. The broker looks up the channel's most recent stored event at subscription time, replays that single event to the new subscriber, and then streams every subsequently published event as it arrives — the same live path any other subscription uses.
Gotchas: if the channel is empty when you subscribe, there's no "last" event to deliver — you simply start receiving new events as they're published, with no error raised. StartFromLast gives you one event, not the last N — if you need a short window of recent history, replay from a sequence number instead. And because "last" is resolved at subscribe time, two subscribers starting a few events apart can each get a different one.
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-last"
private const val CHANNEL = "kotlin-events-store.start-from-last"
fun main() = runBlocking {
val client = KubeMQClient.pubSub {
address = ADDRESS
clientId = CLIENT_ID
}
client.use {
// Publish 5 events
repeat(5) { i ->
client.publishEventStore(eventStoreMessage {
channel = CHANNEL
body = "Event #${i + 1}".toByteArray()
})
println("Published #${i + 1}")
}
delay(500)
// Subscribe with StartFromLast -- receives the last event + any new ones
println("\nSubscribing with StartFromLast:")
val subJob = launch {
client.subscribeToEventsStore {
channel = CHANNEL
startPosition = StartPosition.StartFromLast
}.take(2).collect { msg ->
println(" seq=${msg.sequence}: ${String(msg.body)}")
}
}
delay(500)
// Publish a new event after subscription
client.publishEventStore(eventStoreMessage {
channel = CHANNEL
body = "New event after subscription".toByteArray()
})
subJob.join()
println("Done.")
}
}How It Works
StartPosition.StartFromLastdelivers the last persisted event plus all new events.- This is useful for getting the current state and then staying up-to-date.
- The subscriber receives 2 events: the last pre-existing event (#5) and the new one published after subscription.
Related
Was this page helpful?