Overview

A live Events subscription holds a client-side gRPC streaming call open indefinitely, so a long-running service needs an explicit way to tear one down without closing the whole client connection — for example when a feature flag disables a channel, a worker is draining before shutdown, or a subscription needs to be re-created with different options. Calling subscription.cancel() on the EventsSubscription stops delivery cleanly and frees those resources on both sides.

client.subscribeToEvents(subscription) registers the callback and returns immediately, so the stream keeps running in the background until you cancel it. subscription.cancel() closes the underlying gRPC stream so no further callbacks fire; because that teardown isn't instant, a guard flag like the sample's AtomicBoolean active is a common way to make sure any in-flight callback invocation doesn't process a message after you've logically moved on.

Gotchas: cancel() only affects this one subscription — a consumer group with multiple subscribers keeps delivering to the others. Events already in flight when you call it may still reach the callback briefly after cancellation. And because Events are fire-and-forget, anything published after cancellation reaches the broker but is simply dropped for this subscriber — there's no queue to catch up from later.

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

CancelSubscriptionExample.java

package io.kubemq.example.events;

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

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Cancel Subscription Example for Events
 *
 * Demonstrates how to cancel and manage event subscriptions.
 */
public class CancelSubscriptionExample {

    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-events-cancel-subscription-client";
    private static final String CHANNEL = "java-events.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();

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

        AtomicInteger receivedCount = new AtomicInteger(0);
        AtomicBoolean active = new AtomicBoolean(true);

        EventsSubscription subscription = EventsSubscription.builder()
                .channel(CHANNEL)
                .onReceiveEventCallback(event -> {
                    if (active.get()) {
                        int count = receivedCount.incrementAndGet();
                        System.out.println("  [" + count + "] Received: " + new String(event.getBody()));
                    }
                })
                .onErrorCallback(err -> System.err.println("Error: " + err))
                .build();

        // Subscribe to handle incoming events
        System.out.println("1. Starting subscription...");
        client.subscribeToEvents(subscription);
        Thread.sleep(300);

        // Send messages while subscribed
        System.out.println("2. Sending messages while subscribed...");
        for (int i = 1; i <= 3; i++) {
            client.publishEvent(EventMessage.builder()
                    .id("msg-" + i).channel(CHANNEL)
                    .body(("Message " + i).getBytes()).build());
            Thread.sleep(200);
        }

        // Cancel the subscription (stop receiving new events)
        System.out.println("\n3. Cancelling subscription...");
        active.set(false);
        subscription.cancel();
        System.out.println("   Subscription cancelled.");

        // Send more messages after cancel (subscriber will not receive them)
        System.out.println("4. Sending more messages after cancel...");
        for (int i = 4; i <= 6; i++) {
            client.publishEvent(EventMessage.builder()
                    .id("msg-" + i).channel(CHANNEL)
                    .body(("Message " + i).getBytes()).build());
        }

        Thread.sleep(500);
        System.out.println("5. Received " + receivedCount.get() + " messages (before cancel).");

        // Clean up resources
        client.deleteEventsChannel(CHANNEL);
        client.close();
        System.out.println("\nCancel subscription example completed.");
    }
}

How It Works

EventsSubscription holds the active gRPC streaming call; subscription.cancel() closes the stream so no further callbacks fire.
An AtomicBoolean active flag guards the callback body — messages delivered before the OS-level stream teardown is confirmed are silently dropped.
Messages sent after cancel() reach the broker but no subscriber is registered, demonstrating the fire-and-forget nature of Events.
AtomicInteger receivedCount tracks delivery without locking; the final print confirms exactly how many messages arrived before cancellation.

Cancel Subscription

Overview

Prerequisites

Code

How It Works

Related

On this page