Start at Time Delta
Subscribe to a KubeMQ Events Store channel starting from a relative time offset to replay recent events in Java.
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.
EventsStoreType.StartAtTimeDelta with .eventsStoreSequenceValue(deltaSeconds) 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. The builder reuses eventsStoreSequenceValue for both sequence- and time-based positions, so double-check eventsStoreType is actually set to StartAtTimeDelta. 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 - Java SDK installed (
implementation 'io.kubemq.sdk:kubemq-sdk-Java:3.1.1'(Gradle) or Maven dependency from Getting Started)
Code
package io.kubemq.example.eventsstore;
import io.kubemq.sdk.pubsub.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Start At Time Delta Example
*
* Demonstrates StartAtTimeDelta subscription that replays messages from a relative time.
*/
public class StartAtTimeDeltaExample {
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-eventsstore-start-at-time-delta-client";
private static final String CHANNEL = "java-eventsstore.start-at-time-delta";
public static void main(String[] args) throws InterruptedException {
// Create a client connected to the KubeMQ server
PubSubClient client = PubSubClient.builder().address(ADDRESS).clientId(CLIENT_ID).build();
client.ping();
// Create the events store channel
client.createEventsStoreChannel(CHANNEL);
// Send messages to create history
for (int i = 1; i <= 5; i++) {
client.publishEventStore(EventStoreMessage.builder()
.id("msg-" + i).channel(CHANNEL)
.body(("Event #" + i).getBytes()).build());
Thread.sleep(100);
}
System.out.println("Sent 5 messages.\n");
int deltaSeconds = 60;
AtomicInteger received = new AtomicInteger(0);
CountDownLatch latch = new CountDownLatch(5);
EventsStoreSubscription sub = EventsStoreSubscription.builder()
.channel(CHANNEL)
.eventsStoreType(EventsStoreType.StartAtTimeDelta)
.eventsStoreSequenceValue(deltaSeconds)
.onReceiveEventCallback(e -> {
received.incrementAndGet();
System.out.println(" Received: " + new String(e.getBody()));
latch.countDown();
})
.onErrorCallback(err -> System.err.println("Error: " + err))
.build();
// Subscribe with StartAtTimeDelta (replay from relative time)
client.subscribeToEventsStore(sub);
System.out.println("Subscribed with TimeDelta = " + deltaSeconds + "s.");
latch.await(10, TimeUnit.SECONDS);
System.out.println("\nReceived " + received.get() + " messages from the last " + deltaSeconds + "s.");
// Clean up resources
sub.cancel();
client.deleteEventsStoreChannel(CHANNEL);
client.close();
}
}
How It Works
EventsStoreType.StartAtTimeDeltawith.eventsStoreSequenceValue(deltaSeconds)replays events from the last N seconds relative to now; no explicit timestamp is needed.deltaSeconds = 60means the broker replays all events stored in the last minute; since all five messages were published just before the subscription, all five are replayed.- The delta is passed as an integer via
eventsStoreSequenceValue(reusing the same builder field); the broker interprets it as a seconds offset.
Related
Was this page helpful?