Overview

This tutorial builds a publisher and subscriber on a KubeMQ Events Store channel — reach for this pattern when a subscriber can't guarantee it's listening the instant a message is published. Plain events are fire-and-forget: publish with no one subscribed and the message is gone. Events Store persists every event to a durable, ordered log, so a subscriber connecting seconds or a full restart later still catches up — useful for anything needing a complete history, like an audit trail or event-sourced state.

The two calls involved: client.publishEventStore(message) publishes and returns an EventSendResult confirming storage plus a broker-assigned sequence number, and EventsStoreSubscription.builder() requires an eventsStoreType telling the broker where to start — new events only (StartNewOnly, used here), from the first stored event, or a given sequence or time. Production subscribers usually resume from a saved checkpoint instead of starting fresh.

Gotchas: starting from new events means anything published earlier is silently skipped — this sample papers over that race with a fixed Thread.sleep instead of a ready signal, fine for a demo but not production. Replaying from the first event on every restart replays the whole log, which gets costly on a busy channel. Persistence isn't consumer coordination: each independent subscriber gets its own full replay unless grouped with a consumer group.

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

PersistentPubSubExample.java

package io.kubemq.example.eventsstore;

import io.kubemq.sdk.common.ServerInfo;
import io.kubemq.sdk.pubsub.*;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * Persistent Pub/Sub Example
 *
 * Demonstrates persistent pub/sub with EventsStore -- send and subscribe to stored events.
 */
public class PersistentPubSubExample {

    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-eventsstore-persistent-pubsub-client";
    private static final String CHANNEL = "java-eventsstore.persistent-pubsub";

    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();

        // Verify connection to the server
        ServerInfo info = client.ping();
        System.out.println("Connected to: " + info.getHost());
        // Create the persistent events store channel
        client.createEventsStoreChannel(CHANNEL);

        CountDownLatch latch = new CountDownLatch(3);

        Consumer<EventStoreMessageReceived> onReceive = event -> {
            System.out.println("Received stored event:");
            System.out.println("  ID: " + event.getId());
            System.out.println("  Sequence: " + event.getSequence());
            System.out.println("  Body: " + new String(event.getBody()));
            latch.countDown();
        };

        EventsStoreSubscription subscription = EventsStoreSubscription.builder()
                .channel(CHANNEL)
                .eventsStoreType(EventsStoreType.StartNewOnly)
                .onReceiveEventCallback(onReceive)
                .onErrorCallback(err -> System.err.println("Error: " + err.getMessage()))
                .build();

        // Subscribe to handle incoming stored events
        client.subscribeToEventsStore(subscription);
        System.out.println("Subscribed to events store: " + CHANNEL);
        // Wait for the subscriber to be ready
        Thread.sleep(500);

        // Send persistent event messages
        for (int i = 1; i <= 3; i++) {
            Map<String, String> tags = new HashMap<>();
            tags.put("sequence", String.valueOf(i));

            EventStoreMessage message = EventStoreMessage.builder()
                    .id(UUID.randomUUID().toString())
                    .channel(CHANNEL)
                    .metadata("Persistent event")
                    .body(("Stored event #" + i).getBytes())
                    .tags(tags)
                    .build();

            EventSendResult result = client.publishEventStore(message);
            System.out.println("Sent event #" + i + " (sent=" + result.isSent() + ")");
        }

        // Wait for the subscriber to receive all messages
        latch.await(5, TimeUnit.SECONDS);

        // Clean up resources
        subscription.cancel();
        client.deleteEventsStoreChannel(CHANNEL);
        client.close();
        System.out.println("\nPersistent pub/sub example completed.");
    }
}

// Expected output:
// Connected to: <host>
// Subscribed to events store: java-eventsstore.persistent-pubsub
// Sent event #1 (sent=true)
// Sent event #2 (sent=true)
// Sent event #3 (sent=true)
// Received stored event:
//   ID: <message-id>
//   Sequence: <sequence>
//   Body: Stored event #1
// Received stored event:
//   ID: <message-id>
//   Sequence: <sequence>
//   Body: Stored event #2
// Received stored event:
//   ID: <message-id>
//   Sequence: <sequence>
//   Body: Stored event #3
//
// Persistent pub/sub example completed.

How It Works

client.publishEventStore(message) returns an EventSendResult with a isSent() flag; unlike plain Events, the broker confirms storage before returning.
EventsStoreSubscription.builder() requires an eventsStoreType; StartNewOnly means the subscriber only receives events published after subscription time.
The subscriber is registered and given 500 ms to establish before publishing, ensuring all three stored events arrive in the subscription window.
CountDownLatch coordinates the main thread to wait for all three deliveries before cleanup.

Persistent Pub/Sub

Overview

Prerequisites

Code

How It Works

Related

On this page