Flow Context dan Cancellation

Sebelumnya kita mempelajari operator Flow. Bab ini menjawab dua pertanyaan mendasar tentang eksekusi Flow: di thread mana blok builder dijalankan, dan apa yang terjadi saat scope yang mengumpulkan Flow dibatalkan. Memahami kedua aspek ini penting agar pipeline Flow tetap aman saat berinteraksi dengan I/O berat, dan agar coroutine yang membatalkan diri tidak meninggalkan resource menggantung.

Aturan Context Preservation #

Aturan utama Flow: blok flow { ... } dijalankan di konteks pemanggil collect. Jika collect dipanggil dari Dispatchers.Main, blok builder juga berjalan di Dispatchers.Main. Hal ini disebut context preservation. Konsekuensinya, kode yang melakukan I/O blocking di dalam flow { ... } akan memblokir thread pemanggil — biasanya thread UI atau thread utama.

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

fun main() = runBlocking {
    println("=== 1. Default: flow berjalan di konteks pemanggil ===")
    val flowSederhana = flow {
        println("[flow] thread emit = ${namaPool(Thread.currentThread().name)}")
        emit(1)
    }
    flowSederhana.collect {
        println("[collect] thread = ${namaPool(Thread.currentThread().name)}; nilai=$it")
    }

    println()
    println("=== 2. flowOn(Dispatchers.IO): upstream pindah ke IO ===")
    val flowIO = flow {
        println("[flow] thread emit = ${namaPool(Thread.currentThread().name)}")
        emit("data-1")
        emit("data-2")
    }.flowOn(Dispatchers.IO)

    flowIO.collect {
        println("[collect] thread = ${namaPool(Thread.currentThread().name)}; nilai=$it")
    }

    println()
    println("=== 3. Context preservation: collect tetap di pemanggil ===")
    withContext(Dispatchers.Default) {
        flow {
            println("[flow] thread emit = ${namaPool(Thread.currentThread().name)}")
            emit(42)
        }.flowOn(Dispatchers.IO).collect {
            println("[collect] thread = ${namaPool(Thread.currentThread().name)}; nilai=$it")
        }
    }

    println()
    println("=== 4. Cancellation: collect dibatalkan saat scope dibatalkan ===")
    val hasil = withTimeoutOrNull(60) {
        flow {
            repeat(5) { i ->
                emit(i)
                delay(40)  // total 200ms, lebih dari timeout 60ms
            }
        }.collect { println("[collect] terima $it") }
        "selesai"
    }
    println("[main] hasil collect = $hasil (cancellationException ditangkap denganTimeoutOrNull)")

    println()
    println("=== 5. CancellationException di tengah flow ===")
    val job = launch {
        try {
            flow {
                repeat(10) { i ->
                    emit(i)
                    delay(30)
                }
            }.collect { println("[collect] $it") }
        } catch (e: CancellationException) {
            println("[flow-collect] dibatalkan: ${e::class.simpleName}")
        }
    }
    delay(100)  // biarkan beberapa emit terjadi
    job.cancelAndJoin()
    println("[main] job dibatalkan dan ditunggu")
}

fun namaPool(namaThread: String): String {
    return when {
        namaThread.startsWith("DefaultDispatcher") -> "DefaultDispatcher"
        namaThread.startsWith("DefaultExecutor") -> "DefaultExecutor"
        namaThread.contains("main") -> "main"
        else -> namaThread
    }
}

Output yang Dihasilkan:

=== 1. Default: flow berjalan di konteks pemanggil ===
[flow] thread emit = main
[collect] thread = main; nilai=1

=== 2. flowOn(Dispatchers.IO): upstream pindah ke IO ===
[flow] thread emit = DefaultDispatcher
[collect] thread = main; nilai=data-1
[collect] thread = main; nilai=data-2

=== 3. Context preservation: collect tetap di pemanggil ===
[flow] thread emit = DefaultDispatcher
[collect] thread = DefaultDispatcher; nilai=42

=== 4. Cancellation: collect dibatalkan saat scope dibatalkan ===
[collect] terima 0
[collect] terima 1
[main] hasil collect = null (cancellationException ditangkap denganTimeoutOrNull)

=== 5. CancellationException di tengah flow ===
[collect] 0
[collect] 1
[collect] 2
[collect] 3
[flow-collect] dibatalkan: JobCancellationException
[main] job dibatalkan dan ditunggu

Mari kita bedah bagian-bagian penting dari kode di atas:

1. Bagian 1 (Default): tanpa flowOn, blok flow { ... } berjalan di thread yang sama dengan collect. Output menunjukkan [flow] thread emit = main dan [collect] thread = main. Keduanya di thread main.
2. Bagian 2 (flowOn(Dispatchers.IO)): blok flow { ... } dipindahkan ke thread pool IO. Output membuktikan [flow] thread emit = DefaultDispatcher (Dispatchers.IO memakai thread dari DefaultDispatcher), tetapi collect tetap di main. Inilah inti context preservation: upstream boleh pindah dispatcher, tetapi collect tidak.
3. Bagian 3 (withContext(Dispatchers.Default)): ketika collect dipanggil dari dalam withContext, blok collect ikut pindah ke DefaultDispatcher. Upstream flow { ... } tetap di IO karena flowOn. Output menunjukkan kedua jalur di thread pool yang berbeda.
4. Bagian 4 (withTimeoutOrNull(60)): Flow dipaksa berhenti setelah 60ms. Karena tiap emit ditambah delay(40) memakan lebih dari 60ms, hanya emit pertama (0) dan kedua (1) yang sempat dikumpulkan sebelum CancellationException dilempar. withTimeoutOrNull menangkap exception dan mengembalikan null.

flowOn dan Mekanisme Pengubahan Dispatcher #

Operator flowOn(dispatcher) memindahkan seluruh upstream ke dispatcher baru. Implementasinya: Flow membuat coroutine baru di dispatcher yang ditentukan, lalu mengirim nilai ke collector di konteks pemanggil melalui channel internal. Akibatnya, performa Flow yang sering memanggil flowOn sedikit lebih lambat karena ada context switch, tetapi keuntungannya jauh lebih besar: thread UI tidak terblokir saat Flow melakukan I/O berat.

Cancellation: Flow Menghormati CancellationException #

Flow dirancang untuk cooperative cancellation. Saat scope tempat collect berjalan dibatalkan, collect melempar CancellationException. Blok flow { ... } juga akan berhenti di titik suspend berikutnya (misalnya delay, emit, atau panggilan suspend lain).

Mekanisme ini berarti:

• withTimeout dan withTimeoutOrNull bekerja langsung pada Flow tanpa kode tambahan.
• job.cancel() dari luar menghentikan Flow yang sedang berjalan.
• Resource yang dibuka di dalam flow { ... } sebaiknya ditutup dengan try { ... } finally { ... } agar tetap ditutup saat pembatalan terjadi.

Pada Bagian 5 kode contoh, kita melihat bahwa setelah job.cancelAndJoin() dipanggil, blok catch (e: CancellationException) menangkap exception, mencetak pesan, lalu Flow berakhir dengan rapi. Pola try/catch (CancellationException) berguna untuk log diagnostik atau pembersihan resource sebelum benar-benar berhenti.

Operator Cancellation-Specific #

Beberapa operator Flow berperilaku khusus terkait pembatalan:

Operator cancellable() berguna saat upstream tidak secara native menghormati pembatalan — misalnya, flow { while(true) { emit(...) } } tanpa delay tidak akan merespons cancel() kecuali diberi .cancellable() sebelum collect.

Referensi Lanjutan #

• Flow — panduan flowOn dan penanganan cancellation pada Flow
• Coroutine context and dispatchers — konteks dispatcher untuk Flow

Selanjutnya #

Memahami context dan cancellation melengkapi gambaran teknis Flow. Bab berikutnya, Cold vs Hot Flows, akan membandingkan Flow (cold) dengan SharedFlow dan StateFlow (hot) — kapan memakai satu dan bukan yang lain, serta implikasi desain dari setiap pilihan.