Flow — apa itu, stream cold dan hot dalam coroutine Kotlin

Apa itu Flow di Kotlin?

Flow — adalah tipe dari paket kotlinx.coroutines.flow, yang mewakili aliran data asinkron cold. Pada intinya, Flow adalah urutan coroutine yang mengemisi nilai melalui fungsi emit() dan berakhir baik dengan sukses maupun dengan pengecualian. Koleksi aliran dilakukan melalui operator terminal collect(), yang merupakan fungsi suspend.

Semantik Cold

Cold stream berarti kode di dalam flow-builder dieksekusi ulang untuk setiap pelanggan. Observable.fromIterable di RxJava berperilaku serupa: pelanggan baru menerima semua nilai dari awal. Di Flow, ini diimplementasikan melalui fungsi suspend collect, yang memblokir coroutine selama pengumpulan data.

Flow builders

Kotlin menyediakan beberapa cara untuk membuat Flow: flow { } — konstruksi dasar dengan emit(), flowOf(vararg values) — untuk kumpulan nilai tetap, .asFlow() — ekstensi untuk koleksi dan Sequence. Semua builder bersifat cold — data hanya dihasilkan saat operator terminal dipanggil.

Stream Cold dan Hot

Pembagian menjadi stream cold dan hot adalah konsep kunci pemrograman reaktif. Cold stream (Flow, Observable) memulai pembangkitan data saat berlangganan. Hot stream (Channel, SharedFlow) mengemisi data secara independen — pelanggan hanya menerima apa yang terjadi setelah berlangganan, tanpa awal urutan.

SharedFlow — adalah Flow panas yang dapat memiliki banyak pelanggan dan dapat memutar ulang nilai terakhir saat replay diatur. SharedFlow cocok untuk peristiwa (notifikasi sekali pakai). StateFlow — variannya dengan nilai status tetap, yang menyimpan nilai terakhir untuk pelanggan baru.

ChannelFlow menggunakan Channel di bawah kap, menggabungkan properti Flow dan Channel. Ini mendukung buffering dan backpressure melalui kapasitas (capacity). ChannelFlow berguna saat mengonversi callback-API menjadi aliran reaktif, ketika nilai dikemisi dari coroutine yang berbeda.

Konversi antara cold dan hot

Untuk konversi cold Flow ke hot SharedFlow digunakan operator shareIn(scope, started, replay). Parameter started mengontrol momen mulai: SharingStarted.WhileSubscribed() — aktif selama ada pelanggan, Lazily — mulai pada pelanggan pertama, Eagerly — mulai segera. Konversi sebaliknya — hot ke cold: StateFlow.asFlow() mengembalikan Flow cold yang saat collect mengemisi nilai StateFlow saat ini. Ini nyaman untuk pengujian.

Operator Flow

Flow menyediakan kumpulan operator yang kaya yang bekerja sebagai fungsi suspend di dalam coroutine. Operator tidak memiliki status dan mengembalikan Flow baru — aliran asli tetap tidak berubah. Ini memungkinkan membangun rantai transformasi yang aman tanpa efek samping.

Operator map mentransformasi setiap nilai aliran melalui transformasi asinkron atau sinkron. filter hanya melewatkan nilai yang memenuhi kondisi. catch menangkap pengecualian sebelum operator terminal dan memungkinkan pemulihan aliran. flatMapLatest membatalkan emisi sebelumnya saat nilai baru tiba — mirip dengan switchMap di Rx.

Operator debounce di Flow menunda publikasi nilai dengan batas waktu yang ditentukan. Jika dalam waktu ini datang nilai baru — timer direset. Di Android, debounce digunakan untuk pencarian: permintaan dikirim hanya setelah jeda 300-400 ms, yang mengurangi jumlah panggilan API sebanyak 3-5 kali.

Operator Terminal

Selain collect(), Flow mendukung operator terminal lainnya: toList() mengumpulkan semua nilai ke dalam daftar — berguna untuk pengujian, first() mengembalikan elemen pertama dan membatalkan aliran, single() mengharapkan tepat satu elemen. fold(initial) mengakumulasi nilai melalui fungsi yang diberikan. Semua operator terminal adalah fungsi suspend dan harus dipanggil di dalam coroutine atau fungsi suspend lainnya.

Contoh Kode Flow

Contoh pertama — Flow dasar dengan pembangkitan angka dan transformasi melalui operator map:

val numberFlow = flow {
    for (i in 1..5) {
        delay(500)
        emit(i)
    }
}

scope.launch {
    numberFlow
        .map { "Angka: $it" }
        .collect { value ->
            println(value)
        }
}

Contoh kedua — transformasi aliran dengan filter dan penanganan error melalui catch:

flow {
    emit("data1")
    emit("data2")
    throw RuntimeException("network error")
}
    .catch { e ->
        emit("fallback_data")
    }
    .collect { value ->
        println(value)
    }

Contoh ketiga — penggunaan StateFlow di ViewModel untuk UI reaktif di Jetpack Compose:

class SearchViewModel : ViewModel() {
    private val _query = MutableStateFlow("")
    val results: StateFlow<List<Result>> = _query
        .debounce(300)
        .flatMapLatest { query ->
            repository.search(query)
        }
        .catch { emit(emptyList()) }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

    fun onQueryChanged(query: String) {
        _query.value = query
    }
}

StateFlow dan SharedFlow

StateFlow — adalah Flow panas dengan satu nilai saat ini. Ini menyimpan nilai terakhir dan meneruskannya ke pelanggan baru segera. StateFlow adalah wadah Observable untuk status, mendukung perbandingan equals — jika nilai baru sama dengan nilai saat ini, emisi tidak terjadi. Jetpack Compose menggunakan StateFlow melalui collectAsState().

SharedFlow — adalah Flow panas yang lebih fleksibel tanpa nilai awal wajib. SharedFlow dikonfigurasi melalui replay (jumlah nilai untuk pelanggan baru), extraBufferCapacity (buffer di luar replay), dan onBufferOverflow (strategi saat meluap). SharedFlow ideal untuk peristiwa sekali pakai: navigasi, Snackbar, analitik.

Flow dalam arsitektur Android direkomendasikan oleh Google sebagai sumber data utama (Lapisan: Repository → UseCase → ViewModel). LiveData kalah dalam fleksibilitas: Flow mendukung coroutine, operator, backpressure dan bekerja di luar lapisan UI. Migrasi dari LiveData ke Flow adalah praktik standar dalam proyek Android modern.

Saat menggunakan Flow di ViewModel, pemilihan tipe yang tepat itu penting. StateFlow ideal untuk status UI yang harus bertahan dari rotasi layar. SharedFlow cocok untuk peristiwa di mana pemrosesan ulang tidak dapat diterima — misalnya, navigasi. Flow dengan collect() di lifecycleScope memberikan kontrol maksimal atas konteks eksekusi, tetapi memerlukan pembatalan manual saat meninggalkan layar.

Pengujian Flow dilakukan melalui kotlinx-coroutines-test. Pustaka ini menyediakan TestDispatcher — waktu virtual yang memungkinkan mempercepat penundaan (delay) dan mengontrol urutan eksekusi coroutine. TestScope.runTest { } menciptakan lingkungan terisolasi untuk menguji Flow. Operator toList() sering digunakan dalam pengujian untuk mengumpulkan semua nilai flow dengan timeout, untuk memeriksa apakah aliran telah mengemisi urutan data yang benar.

Flow terintegrasi dengan baik dengan Room (pustaka Android untuk database): metode DAO dapat mengembalikan Flow<List<Entity>>. Room secara otomatis mengemisi nilai baru saat ada perubahan tabel — UI diperbarui tanpa pemicu manual. Ini diimplementasikan melalui InvalidationTracker, yang di bawah kap menggunakan Flow dengan callbackFlow. Pendekatan semacam itu menghilangkan kebutuhan akan LiveData dan membuat lapisan data sepenuhnya berorientasi coroutine. Jetpack Compose melalui collectAsState() berlangganan ke StateFlow dan hanya menggambar ulang komponen yang datanya telah berubah — ini memberikan performa yang tidak dapat dicapai dengan arsitektur berorientasi LiveData. DataStore (pengganti SharedPreferences) juga mengembalikan Flow<Preferences>, memastikan pembacaan reaktif pengaturan aplikasi tanpa pemicu pembaruan manual.

Flow mendukung komunikasi antarproses melalui kotlinx-coroutines-core di JVM tanpa pustaka tambahan. Misalnya, dalam aplikasi server di Ktor, Flow dapat mewakili aliran pesan WebSocket yang masuk. Setiap pesan dikemisi ke aliran, melewati filter dan agregasi melalui operator, dan hasilnya dikirim ke klien. Pendekatan semacam itu menggantikan pustaka reaktif seperti Reactor atau RxJava dalam proyek Kotlin.

Kompatibilitas Flow dengan kode RxJava yang ada disediakan oleh modul kotlinx-coroutines-rx3. Fungsi ekstensi Flow.asObservable() mengonversi Flow ke Observable dari RxJava 3. Konversi sebaliknya — CompletableSource.asFlow(), Observable.asFlow(). Ini menyederhanakan migrasi dari RxJava ke coroutine: proyek dapat ditulis ulang secara bertahap, meninggalkan sebagian lapisan pada RxJava. Saat konversi, perlu diperhatikan perbedaan semantik cold/hot: Observable bisa cold maupun hot, Flow selalu cold untuk Flow biasa dan hot untuk SharedFlow.

Penanganan Error dan Pengujian Flow

Untuk penanganan error di Flow ada kekhasan: jika pengecualian terjadi di dalam flow-builder sebelum operator terminal, itu diteruskan ke catch. Jika pengecualian terjadi di operator setelah builder, catch setelah operator ini menangkapnya. retryWhen memungkinkan mengulangi langganan dengan kondisi: ulangi pada error jaringan hingga 3 kali, tetapi jangan ulangi pada CancellationException. Flow menghilangkan error yang bergantung pada status karena tidak menyimpan status — ini menyederhanakan debugging dibandingkan dengan Observable, di mana Subject menyimpan status internal.

Pengujian Flow dengan kotlinx-coroutines-test menggunakan TestDispatcher untuk simulasi penundaan. Turbine — pustaka populer dari komunitas untuk menguji Flow: test { } menjalankan Flow, awaitItem() menunggu nilai berikutnya, awaitComplete() menunggu selesai. Turbine menambahkan timeout default, yang mencegah pengujian menggantung. Untuk menguji StateFlow gunakan .testIn(scope) dengan pemeriksaan nilai dalam urutan kronologis.