Flow Operators

Bab sebelumnya memperkenalkan Flow sebagai cold stream yang baru dijalankan saat di-collect. Sekarang kita akan menyusun pipeline transformasi di atas Flow menggunakan operator. Operator Flow mirip dengan operator pada Sequence atau Collection, tetapi setiap langkah mendukung pemanggilan suspend. Kita akan membahas dua kelompok operator: operator sekuensial seperti map, filter, dan transform; lalu operator kombinasi dan konkurensi seperti zip, combine, flatMapConcat, flatMapMerge, dan flatMapLatest.

Operator Transformasi Dasar #

Tiga operator paling sering dipakai adalah map, filter, dan transform. Operator map menerapkan fungsi satu-ke-satu, filter menyaring nilai berdasarkan predikat, dan transform memberikan kebebasan untuk memanggil emit(...) sebanyak yang dibutuhkan.

kotlin

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

@OptIn(kotlinx.coroutines.ExperimentalCoroutinesApi::class, kotlinx.coroutines.FlowPreview::class)
fun main() = runBlocking {
    println("=== 1. map, filter, transform ===")
    flowOf(1, 2, 3, 4, 5)
        .filter { it % 2 == 1 }
        .map { it * 10 }
        .collect { println("hasil: $it") }

    println()
    println("=== 2. transform fleksibel (emit banyak per input) ===")
    flowOf("A", "B", "C")
        .transform {
            emit(it.lowercase())
            emit("---")
        }
        .collect { println("baris: $it") }

    println()
    println("=== 3. take dan drop ===")
    val ambil3 = (1..10).asFlow().take(3).toList()
    val lewati2 = (1..5).asFlow().drop(2).toList()
    println("take(3) dari 1..10 -> $ambil3")
    println("drop(2) dari 1..5  -> $lewati2")

    println()
    println("=== 4. zip: pasangan dua flow ===")
    val nama = flowOf("Andi", "Budi", "Citra")
    val umur = flowOf(20, 21, 22)
    nama.zip(umur) { n, u -> "$n berumur $u" }
        .collect { println("zip: $it") }

    println()
    println("=== 5. combine: nilai terbaru dari kedua flow ===")
    val sinyalA = flowOf(1, 2, 3).onEach { delay(50) }
    val sinyalB = flowOf("x", "y").onEach { delay(80) }
    sinyalA.combine(sinyalB) { a, b -> "($a,$b)" }
        .collect { println("combine: $it") }

    println()
    println("=== 6. flatMapConcat, flatMapMerge, flatMapLatest ===")
    println("-- flatMapConcat (urut) --")
    (1..3).asFlow()
        .flatMapConcat { i -> flowOf(i, i * 10) }
        .collect { println("concat: $it") }

    println("-- flatMapMerge (paralel) --")
    (1..3).asFlow()
        .flatMapMerge { i -> flowOf(i, i * 10) }
        .collect { println("merge: $it") }

    println("-- flatMapLatest (hanya yang terbaru) --")
    (1..3).asFlow().onEach { delay(60) }
        .flatMapLatest { i -> flow {
            emit("mulai-$i")
            delay(80)
            emit("selesai-$i")
        } }
        .collect { println("latest: $it") }

    println()
    println("=== 7. debounce dan distinctUntilChanged ===")
    println("-- debounce 70ms --")
    flow {
        emit("a"); delay(20)
        emit("b"); delay(20)
        emit("c"); delay(120)  // c lewat karena jeda > 70ms
        emit("d"); delay(120)
    }.debounce(70)
        .collect { println("debounce lewat: $it") }

    println("-- distinctUntilChanged --")
    flowOf(1, 1, 2, 2, 3, 3, 1, 1)
        .distinctUntilChanged()
        .collect { println("distinct: $it") }
}

Output yang Dihasilkan:

text
=== 1. map, filter, transform ===
hasil: 10
hasil: 30
hasil: 50

=== 2. transform fleksibel (emit banyak per input) ===
baris: a
baris: ---
baris: b
baris: ---
baris: c
baris: ---

=== 3. take dan drop ===
take(3) dari 1..10 -> [1, 2, 3]
drop(2) dari 1..5  -> [3, 4, 5]

=== 4. zip: pasangan dua flow ===
zip: Andi berumur 20
zip: Budi berumur 21
zip: Citra berumur 22

=== 5. combine: nilai terbaru dari kedua flow ===
combine: (1,x)
combine: (2,x)
combine: (3,x)
combine: (3,y)

=== 6. flatMapConcat, flatMapMerge, flatMapLatest ===
-- flatMapConcat (urut) --
concat: 1
concat: 10
concat: 2
concat: 20
concat: 3
concat: 30
-- flatMapMerge (paralel) --
merge: 1
merge: 10
merge: 2
merge: 20
merge: 3
merge: 30
-- flatMapLatest (hanya yang terbaru) --
latest: mulai-1
latest: mulai-2
latest: mulai-3
latest: selesai-3

=== 7. debounce dan distinctUntilChanged ===
-- debounce 70ms --
debounce lewat: c
debounce lewat: d
-- distinctUntilChanged --
distinct: 1
distinct: 2
distinct: 3
distinct: 1

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

filter { it % 2 == 1 }.map { it * 10 }: pipeline saring-ubah. Filter membuang nilai genap; map mengalikan nilai ganjil dengan 10. Hasilnya 10, 30, 50 dari input 1, 3, 5.
transform { emit(...); emit(...) }: operator paling fleksibel. Setiap input dapat memicu nol, satu, atau banyak emit. Pada contoh, satu huruf memicu dua emit: huruf kecil dan string pemisah "---".
take(3) dan drop(2): take membatasi flow hanya pada n nilai pertama lalu membatalkan upstream. drop melewatkan n nilai pertama dan meneruskan sisanya. Output membuktikan urutan tetap dipertahankan.
zip(umur) { n, u -> ... }: memasangkan dua flow secara sekuensial. Zip menunggu kedua flow emit, lalu memanggil transformasi dengan pasangan nilai pada indeks yang sama. Jika salah satu flow habis, zip berakhir.
combine(sinyalB) { a, b -> ... }: berbeda dengan zip, combine memakai nilai terbaru dari kedua flow. Saat sinyalA emit 3 dan sinyalB masih "x", kombinasi (3,x) dihasilkan; saat sinyalB akhirnya emit "y", kombinasi (3,y) dihasilkan.

flatMapConcat vs flatMapMerge vs flatMapLatest #

Tiga operator flatMap* ini memetakan setiap nilai input ke Flow baru, lalu menggabungkan hasilnya. Perbedaannya pada strategi konkurensi:

Operator	Strategi	Cocok untuk
`flatMapConcat`	Urut — flow berikutnya menunggu flow saat ini selesai	Pipeline yang harus berurutan
`flatMapMerge`	Paralel — semua flow berjalan bersamaan, hasil dikumpulkan saat siap	Fan-out ke banyak sumber independen
`flatMapLatest`	Hanya flow terbaru — saat input baru datang, flow lama dibatalkan	Pencarian real-time, ketik-sambil-cari

Pada output flatMapLatest, kita melihat mulai-1 langsung diikuti mulai-2 dan mulai-3 — flow untuk input 1 dan 2 dibatalkan sebelum sempat emit selesai- karena input baru datang lebih cepat daripada delay(80) di dalam flow.

debounce dan distinctUntilChanged #

Operator debounce(millis) membuang nilai yang muncul berurutan terlalu cepat, dan hanya meneruskan nilai setelah tidak ada emit baru selama millis milidetik. Operator ini cocok untuk search-as-you-type: kita tidak ingin memproses setiap ketikan, cukup saat pengguna berhenti mengetik. Pada contoh, "a" dan "b" dibuang karena jeda antar-emit hanya 20ms (kurang dari 70ms); "c" dan "d" diteruskan karena setelahnya ada jeda 120ms.

distinctUntilChanged() membuang nilai yang sama dengan nilai sebelumnya. Pada contoh, deretan 1,1,2,2,3,3,1,1 menjadi 1,2,3,1 — hanya perubahan nilai yang diteruskan. Operator ini berguna untuk mencegah pemrosesan ulang state yang identik.

Operator Pendukung Lain #

Beberapa operator pendukung yang sering muncul bersama operator di atas:

Operator	Kegunaan
`onEach { }`	Side-effect per nilai tanpa mengubah nilai
`onStart { }`	Dipanggil sebelum flow mulai emit
`onCompletion { }`	Dipanggil saat flow selesai (sukses, gagal, atau dibatalkan)
`catch { }`	Menangkap exception dari upstream; tidak menangkap exception dari `collect`
`flowOn(Dispatchers.IO)`	Mengubah dispatcher eksekusi upstream (dibahas di bab berikutnya)
`conflate()`	Hanya mempertahankan nilai terbaru jika collector lambat

Referensi Lanjutan #

Flow — panduan resmi operator map, filter, flatMapMerge, combine
kotlinx.coroutines core — API reference operator Flow

Selanjutnya #

Operator Flow membuka kemungkinan pipeline fungsional yang asynchronous. Bab berikutnya, Flow Context dan Cancellation, akan menjawab pertanyaan: di thread mana blok flow { ... } dijalankan? Bagaimana memindahkannya ke Dispatchers.IO? Dan bagaimana pembatalan scope memengaruhi flow yang sedang berjalan?