Start from First
Subscribe and replay all events from the beginning
Overview
A new consumer joining an Events Store channel usually needs more than what happens next — it needs everything that already happened. EventsStoreType.StartFromFirst solves that by replaying the channel's complete stored history before switching to live delivery, so a service can rebuild its state from scratch instead of starting with a blank slate and hoping nothing important was missed.
Under the hood, the broker walks the store from the oldest retained sequence forward, streaming each event to your onReceiveEventCallback in order, then hands off to live delivery of new events without a gap. You don't manage offsets or checkpoints yourself — the start position is set once, at subscription time, via .eventsStoreType(EventsStoreType.StartFromFirst).
Gotchas: on a long-lived channel this can mean replaying millions of events before anything new shows up, so it's the wrong choice for a consumer that only cares about "from now on" (use StartNewOnly for that). Retention and expiration policies still apply — events already purged by TTL or max-count limits are gone and won't be replayed, so "full history" only means what the store still has.
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;
/**
* StartFromFirst Example
*
* Demonstrates subscribing with StartFromFirst to replay all messages from the beginning.
*/
public class StartFromFirstExample {
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-eventsstore-start-from-first-client";
private static final String CHANNEL = "java-eventsstore.start-from-first";
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 before subscribing (to create history)
for (int i = 1; i <= 5; i++) {
client.publishEventStore(EventStoreMessage.builder()
.id("msg-" + i).channel(CHANNEL)
.body(("Message #" + i).getBytes()).build());
}
System.out.println("Sent 5 messages.\n");
AtomicInteger received = new AtomicInteger(0);
CountDownLatch latch = new CountDownLatch(5);
EventsStoreSubscription sub = EventsStoreSubscription.builder()
.channel(CHANNEL)
.eventsStoreType(EventsStoreType.StartFromFirst)
.onReceiveEventCallback(e -> {
int n = received.incrementAndGet();
System.out.println(" [" + n + "] Seq " + e.getSequence() + ": " + new String(e.getBody()));
latch.countDown();
})
.onErrorCallback(err -> System.err.println("Error: " + err))
.build();
// Subscribe with StartFromFirst to replay all messages from the beginning
client.subscribeToEventsStore(sub);
System.out.println("Subscribed with StartFromFirst.");
// Wait for all historical messages to be received
latch.await(10, TimeUnit.SECONDS);
System.out.println("\nReceived " + received.get() + " messages (complete history).");
// Clean up resources
sub.cancel();
client.deleteEventsStoreChannel(CHANNEL);
client.close();
}
}
How It Works
- Five messages are published before the subscription is created; with
StartFromFirstthe broker immediately streams the full stored history to the new subscriber. - The callback prints the server-assigned
e.getSequence()for each event, confirming the replay order starts at sequence 1. CountDownLatch(5)ensures the main thread waits for all historical events before cleanup — useful when the store is large and replay takes a moment.
Related
Was this page helpful?