Cancel Subscription
Cancel an active subscription to a KubeMQ Events Store channel and stop receiving persisted events in the Java SDK.
Overview
Every events store subscription opens a long-lived stream to the broker — a background gRPC stream that keeps pulling delivered events until you tell it to stop. Calling cancel() on the EventsStoreSubscription is how you release that stream deliberately: shutting down a worker, rotating consumers, or tearing down a test without leaking connections or leaving a dangling stream on the server.
Internally, sub.cancel() sends a close signal that unwinds the receive loop and closes the underlying gRPC stream; the client detaches from the broker-side subscription registration so no further callbacks fire afterward.
Gotchas: cancelling only stops this subscription — the channel keeps storing every event published afterward, so nothing is lost, and a fresh subscription with StartFromFirst (or a specific sequence) picks up exactly where this one left off. Cancellation isn't instantaneous; a callback already in flight when you call cancel() may still complete, so don't assume zero more invocations the instant the call returns.
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;
/**
* Cancel Subscription Example (EventsStore)
*
* Demonstrates cancelling EventsStore subscriptions.
*/
public class CancelSubscriptionExample {
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-eventsstore-cancel-subscription-client";
private static final String CHANNEL = "java-eventsstore.cancel-subscription";
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 the store before subscribing
for (int i = 1; i <= 5; i++) {
client.publishEventStore(EventStoreMessage.builder()
.id("store-" + i).channel(CHANNEL)
.body(("Stored message " + i).getBytes()).build());
}
System.out.println("Sent 5 messages to store.\n");
AtomicInteger received = new AtomicInteger(0);
CountDownLatch cancelPoint = new CountDownLatch(1);
EventsStoreSubscription sub = EventsStoreSubscription.builder()
.channel(CHANNEL)
.eventsStoreType(EventsStoreType.StartFromFirst)
.onReceiveEventCallback(e -> {
int n = received.incrementAndGet();
System.out.println(" [" + n + "] " + new String(e.getBody()));
if (n >= 3) cancelPoint.countDown();
})
.onErrorCallback(err -> System.err.println("Error: " + err))
.build();
// Subscribe to handle incoming stored events
System.out.println("Subscribing (will cancel after 3 messages)...\n");
client.subscribeToEventsStore(sub);
// Wait until we receive 3 messages, then cancel
cancelPoint.await(5, TimeUnit.SECONDS);
Thread.sleep(100);
// Cancel the subscription (stop receiving)
sub.cancel();
System.out.println("\nSubscription cancelled. Received " + received.get() + " of 5 messages.");
// Clean up resources
client.deleteEventsStoreChannel(CHANNEL);
client.close();
}
}
How It Works
StartFromFirstreplays all stored events from sequence 1 when the subscription is created, so the subscriber immediately receives the five pre-published messages.- A
CountDownLatch cancelPointcounts down when the third event arrives, signalling the main thread to callsub.cancel()and close the gRPC stream. - Because storage is server-side, the remaining two stored messages are not lost — they can be replayed by a new subscriber. Only this subscription stops receiving.
Related
Was this page helpful?