# Cancel Subscription (/sdks/kotlin/how-to/events-store/cancel-subscription)



## Overview [#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 [#prerequisites]

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

## Code [#code]

```kotlin title="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 [#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.

## Related [#related]

* [Persistent Pub/Sub](/sdks/kotlin/tutorials/persistent-pubsub)
* [Events Cancel Subscription](/sdks/kotlin/how-to/events/cancel-subscription)
* [Events Store Reference](/sdks/kotlin/reference/events-store)