Start from Last
Subscribe to a KubeMQ Events Store channel starting from the most recently stored event using the Java SDK.
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. EventsStoreType.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, EventsStoreType.StartFromLast is set on the events store subscription options. 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 - 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.common.ServerInfo;
import io.kubemq.sdk.pubsub.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
* StartFromLast Example
*
* Demonstrates subscribing with StartFromLast which receives the most recent
* stored message plus all new ones.
*/
public class StartFromLastExample {
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-eventsstore-start-from-last-client";
private static final String CHANNEL = "java-eventsstore.start-from-last";
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 historical messages before subscribing
for (int i = 1; i <= 5; i++) {
client.publishEventStore(EventStoreMessage.builder()
.id("hist-" + i).channel(CHANNEL)
.body(("Historical #" + i).getBytes()).build());
}
System.out.println("Sent 5 historical messages.\n");
AtomicInteger received = new AtomicInteger(0);
CountDownLatch latch = new CountDownLatch(3);
EventsStoreSubscription sub = EventsStoreSubscription.builder()
.channel(CHANNEL)
.eventsStoreType(EventsStoreType.StartFromLast)
.onReceiveEventCallback(e -> {
int n = received.incrementAndGet();
String label = (n == 1) ? "LAST" : "NEW " + (n - 1);
System.out.println(" [" + label + "] " + new String(e.getBody()));
latch.countDown();
})
.onErrorCallback(err -> System.err.println("Error: " + err))
.build();
// Subscribe with StartFromLast (most recent + new messages only)
client.subscribeToEventsStore(sub);
Thread.sleep(500);
// Send new messages after subscribing
for (int i = 1; i <= 2; i++) {
client.publishEventStore(EventStoreMessage.builder()
.id("new-" + i).channel(CHANNEL)
.body(("New message #" + i).getBytes()).build());
Thread.sleep(200);
}
latch.await(5, TimeUnit.SECONDS);
System.out.println("\nReceived " + received.get() + " (1 last + 2 new).");
// Clean up resources
sub.cancel();
client.deleteEventsStoreChannel(CHANNEL);
client.close();
}
}
How It Works
StartFromLastdelivers exactly one historical event — the most recent stored message — then continues streaming new events as they are published.- Five historical messages are sent before subscribing; only the last one (
Historical #5) is replayed on subscription. - Two new messages are published after the subscription is active; the latch counts to 3 (1 last + 2 new) to confirm both delivery modes work.
Related
Was this page helpful?