FlowCancellable.kt

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

suspend fun main() {
    coroutineScope {
        val sequence = generateSequence(0) { it + 1 }
            .asFlow()
            .take(5)
            .onEach {
                // purposefully non-cancellable sleep
                withContext(NonCancellable) {
                    delay(1000)
                }
            }

        val nonCancelJob = launch {
            sequence
                .collect {
                    System.err.println("Non-cancellable: $it")
                }
        }
        val cancelJob = launch {
            sequence
                .cancellable()
                .collect {
                    System.err.println("Cancellable: $it")
                }
        }
        delay(2000)
        nonCancelJob.cancel()
        cancelJob.cancel()
    }
}