Cancel Subscription
Cancel an active event subscription using coroutine cancellation
Overview
A live Events subscription holds a coroutine collecting a Flow 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. Cancelling the coroutine Job that's collecting the subscription stops delivery cleanly using ordinary Kotlin structured concurrency, rather than a bespoke unsubscribe API.
client.subscribeToEvents { ... } returns a Flow that's collected inside a coroutine launched with launch; that launch call gives you back a Job. Calling subJob.cancel() stops the Flow collection and closes the subscription, and because structured concurrency propagates cancellation automatically, any resources tied to that coroutine's scope are cleaned up without extra bookkeeping.
Gotchas: cancelling the job only affects this one subscription — other subscribers on the same channel keep receiving events. Messages already in flight when you cancel may still be collected before the Flow notices. And because Events are fire-and-forget, anything published after cancellation is simply dropped for this subscriber — there's no queue to catch up from later.
Prerequisites
- KubeMQ server running on
localhost:50000 - Kotlin SDK installed (
implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))
Code
package io.kubemq.sdk.examples.events
import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.pubsub.eventMessage
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-cancel-subscription"
private const val CHANNEL = "kotlin-events.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 subscription...")
val subJob: Job = launch {
client.subscribeToEvents {
channel = CHANNEL
}.collect { msg ->
val count = receivedCount.incrementAndGet()
println(" [$count] Received: ${String(msg.body)}")
}
}
delay(300)
// 2. Send messages while subscribed
println("2. Sending messages while subscribed...")
repeat(3) { i ->
client.publishEvent(eventMessage {
channel = CHANNEL
body = "Message ${i + 1}".toByteArray()
})
delay(200)
}
delay(500)
// 3. Cancel the subscription
println("\n3. Cancelling subscription...")
subJob.cancel()
println(" Subscription cancelled.")
// 4. Send more messages after cancel (subscriber will not receive them)
println("4. Sending more messages after cancel...")
repeat(3) { i ->
client.publishEvent(eventMessage {
channel = CHANNEL
body = "Message ${i + 4}".toByteArray()
})
}
delay(500)
println("5. Received ${receivedCount.get()} messages (before cancel).")
println("\nCancel subscription example completed.")
}
}How It Works
- Subscriptions in the Kotlin SDK are
Flow-based, collected inside a coroutine. - Cancelling the coroutine job (
subJob.cancel()) stops the Flow collection and closes the subscription. - Messages published after cancellation are not received by the cancelled subscriber.
- This is idiomatic Kotlin -- coroutine structured concurrency handles cleanup automatically.
Related
Was this page helpful?