Overview

Every events store subscription opens a long-lived stream to the broker — a coroutine Job collecting a Flow of delivered events until you tell it to stop. Cancelling that Job is how you release the stream deliberately: shutting down a worker, rotating consumers, or tearing down a coroutine scope without leaking connections or leaving a dangling stream on the server.

Internally, subJob.cancel() stops the coroutine collecting subscribeToEventsStore's Flow, which propagates the cancellation down to the underlying stream and detaches from the broker-side subscription registration.

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 StartAtSequence picks up exactly where this one left off. Coroutine cancellation is cooperative — if a collector is mid-callback when cancel() is called, that invocation still completes before the Flow actually stops.

Prerequisites

KubeMQ server running on localhost:50000
Kotlin SDK installed (implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))

Code

CancelSubscriptionExample.kt

package io.kubemq.sdk.examples.eventsstore

import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.pubsub.StartPosition
import io.kubemq.sdk.pubsub.eventStoreMessage
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.util.concurrent.atomic.AtomicInteger

private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-events-store-cancel-subscription"
private const val CHANNEL = "kotlin-events-store.cancel-subscription"

fun main() = runBlocking {
    val client = KubeMQClient.pubSub {
        address = ADDRESS
        clientId = CLIENT_ID
    }

    client.use {
        val receivedCount = AtomicInteger(0)

        // 1. Start subscription
        println("1. Starting events store subscription...")
        val subJob: Job = launch {
            client.subscribeToEventsStore {
                channel = CHANNEL
                startPosition = StartPosition.StartNewOnly
            }.collect { msg ->
                val count = receivedCount.incrementAndGet()
                println("  [$count] Received: seq=${msg.sequence} ${String(msg.body)}")
            }
        }

        delay(300)

        // 2. Send messages while subscribed
        println("2. Sending persistent events while subscribed...")
        repeat(3) { i ->
            client.publishEventStore(eventStoreMessage {
                channel = CHANNEL
                body = "Message ${i + 1}".toByteArray()
            })
            delay(200)
        }

        delay(500)

        // 3. Cancel subscription
        println("\n3. Cancelling subscription...")
        subJob.cancel()
        println("   Subscription cancelled.")

        // 4. Send more messages (should not be received)
        println("4. Sending more messages after cancel...")
        repeat(3) { i ->
            client.publishEventStore(eventStoreMessage {
                channel = CHANNEL
                body = "Post-cancel message ${i + 1}".toByteArray()
            })
        }

        delay(500)
        println("5. Received ${receivedCount.get()} messages (before cancel).")
        println("\nCancel subscription example completed.")
    }
}

How It Works

The subscription runs in a coroutine job; cancelling the job stops the Flow collection.
Post-cancel messages are still persisted but not delivered to the cancelled subscriber.
Re-subscribing with StartFromFirst or StartAtSequence can replay missed messages.

Cancel Subscription

Overview

Prerequisites

Code

How It Works

On this page

Cancel Subscription

Overview

Prerequisites

Code

How It Works

Related

On this page