Start at Time Delta

Overview

A time-delta subscription starts replay from a relative offset — "the last 60 seconds" — instead of a fixed timestamp or sequence number. It's the right tool when a consumer knows how long it was offline but not the exact moment it disconnected: a worker restarting after a deploy, a dashboard reconnecting after a blip, or a batch job that only cares about "recent" history. Computing an absolute cutoff yourself is bookkeeping the broker can do for you.

StartPosition.StartAtTimeDelta(60) passes the offset to the broker, which resolves it to now - delta at subscription time, replays every stored event from that point forward, then hands off to live delivery — the same replay-to-live transition as an absolute-time or sequence-based start.

Gotchas: the delta is evaluated once, server-side, at subscription creation — it does not "slide" as time passes. A delta of zero replays nothing and behaves like starting from new events only. And since the window is wall-clock based, clock skew between producers and the broker can shift which events land inside or outside the boundary.

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-at-time-delta"
private const val CHANNEL = "kotlin-events-store.start-at-time-delta"

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 = "Delta event #${i + 1}".toByteArray()
            })
            println("Published #${i + 1}")
        }

        delay(500)

        // Subscribe with StartAtTimeDelta(60) -- events from the last 60 seconds
        println("\nSubscribing with StartAtTimeDelta(60s):")
        val subJob = launch {
            client.subscribeToEventsStore {
                channel = CHANNEL
                startPosition = StartPosition.StartAtTimeDelta(60)
            }.take(5).collect { msg ->
                println("  seq=${msg.sequence}: ${String(msg.body)}")
            }
        }

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

How It Works

• StartPosition.StartAtTimeDelta(60) replays events from the last 60 seconds.
• The delta is calculated server-side relative to the current server time.
• This is simpler than StartAtTime when you need a relative offset rather than an absolute timestamp.
• Useful for "catch up on recent events" scenarios without tracking exact timestamps.

Related

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