KAL
Menu

Flow

Cold stream, flow builder, operator, dan collect.

Daftar isi

Flow adalah aliran nilai asynchronous yang dingin (cold stream), yaitu rangkaian nilai nol hingga banyak yang diproduksi satu per satu setelah alirannya dikumpulkan. Sebuah fungsi suspend mengirim satu nilai, sedangkan Flow mengirim banyak nilai secara berurutan dengan gaya yang sama. Topik ini membahas builder flow dan operator terminal collect, sifat cold stream beserta buktinya, serta operator antara seperti filter dan map yang disusun berantai.

Builder flow dan collect#

Aliran dibuat dengan builder flow, yaitu fungsi yang isinya memanggil emit untuk setiap nilai yang dikirim ke aliran. Setiap pemanggilan emit boleh dijeda, misalnya oleh delay atau pekerjaan jaringan, sehingga produksi nilai dapat berjalan sambil menunggu. Fungsi yang mengembalikan Flow biasanya dideklarasikan tanpa suspend karena builder-nya baru dijalankan saat aliran dikumpulkan.

kotlin main.kt 
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun bacaSensor(): Flow<Int> = flow {
    for (suhu in listOf(21, 22, 23)) {
        emit(suhu)
    }
}

fun main() = runBlocking {
    bacaSensor().collect { suhu ->
        println("Suhu terbaca: $suhu")
    }
}
bash 
# compiler Kotlin dan library coroutines yang dipakai jalur ini
export PATH=/home/kal/my-project/tools/kotlinc/bin:/home/kal/my-project/tools/jdk-merged/bin:$PATH
export LIBS=/home/kal/my-project/tools/libs
export STDLIB=/home/kal/my-project/tools/kotlinc/lib/kotlin-stdlib.jar
# kompilasi main.kt dengan library coroutines pada classpath, lalu jalankan programnya
kotlinc main.kt -cp "$LIBS/kotlinx-coroutines-core-jvm-1.11.0.jar" -d out.jar && java -cp "out.jar:$LIBS/kotlinx-coroutines-core-jvm-1.11.0.jar:$STDLIB" MainKt
text Keluaran 
Suhu terbaca: 21
Suhu terbaca: 22
Suhu terbaca: 23

Fungsi collect adalah operator terminal, yaitu operator yang benar-benar menyalakan aliran dan menerima setiap nilai di dalam lambda pemakainya. Tanpa operator terminal, tidak ada satu baris builder pun yang berjalan, karena Flow menyimpan cara memproduksi nilai dan belum menjalankannya. Nilai juga diproses satu per satu secara berurutan, sebagaimana keluarannya yang tercetak sesuai urutan emit.

Flow sebagai cold stream#

Flow bersifat dingin, yang berarti builder-nya dieksekusi ulang dari awal untuk setiap pengumpulan. Aliran tidak menyimpan nilai maupun hasil sebelumnya, melainkan menyimpan resepnya, dan setiap pihak yang mengumpulkan mendapat eksekusi resep yang baru. Sifat ini dibuktikan dengan mencetak sesuatu di dalam builder, lalu mengumpulkan aliran yang sama dua kali.

kotlin dingin.kt 
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun buatAngka(): Flow<Int> = flow {
    println("Builder dieksekusi")
    emit(1)
    emit(2)
}

fun main() = runBlocking {
    println("Collect pertama dimulai")
    buatAngka().collect { nilai -> println("Nilai: $nilai") }
    println("Collect kedua dimulai")
    buatAngka().collect { nilai -> println("Nilai: $nilai") }
}
bash 
# kompilasi dingin.kt dengan library coroutines pada classpath, lalu jalankan programnya
kotlinc dingin.kt -cp "$LIBS/kotlinx-coroutines-core-jvm-1.11.0.jar" -d dingin.jar && java -cp "dingin.jar:$LIBS/kotlinx-coroutines-core-jvm-1.11.0.jar:$STDLIB" DinginKt
text Keluaran 
Collect pertama dimulai
Builder dieksekusi
Nilai: 1
Nilai: 2
Collect kedua dimulai
Builder dieksekusi
Nilai: 1
Nilai: 2

Baris Builder dieksekusi tercetak dua kali, masing-masing tepat setelah collect dimulai, bukan saat fungsi buatAngka dipanggil. Urutan itu memperlihatkan dua sifat sekaligus, yaitu aliran baru berjalan ketika dikumpulkan dan setiap pengumpulan mendapat eksekusi penuh yang baru. Konsekuensinya, dua pengumpulan pada aliran dingin tidak berbagi keadaan apa pun, sehingga aliran semacam ini cocok untuk membaca ulang sumber data, bukan untuk dipakai berbagi kejadian antar banyak penerima.

Operator antara sumber dan collect#

Operator antara dipasang di antara sumber dan operator terminal, dan setiap operator mengembalikan Flow baru yang masih dingin. Operator filter meloloskan nilai tertentu, sedangkan map mengubah setiap nilai menjadi bentuk lain, dan keduanya dapat dirangkai sebagaimana operator collection atau Sequence. Aliran dirangkai berantai pada program berikut, yang menyaring pembacaan sensor lalu mengubah nilainya menjadi teks.

kotlin operasi.kt 
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val suhu = flowOf(21, 30, 19, 28, 35)
    suhu
        .filter { it > 25 }
        .map { "suhu $it derajat" }
        .collect { println("Terdeteksi $it") }
}
bash 
# kompilasi operasi.kt dengan library coroutines pada classpath, lalu jalankan programnya
kotlinc operasi.kt -cp "$LIBS/kotlinx-coroutines-core-jvm-1.11.0.jar" -d operasi.jar && java -cp "operasi.jar:$LIBS/kotlinx-coroutines-core-jvm-1.11.0.jar:$STDLIB" OperasiKt
text Keluaran 
Terdeteksi suhu 30 derajat
Terdeteksi suhu 28 derajat
Terdeteksi suhu 35 derajat

Dari lima pembacaan, hanya tiga yang lolos ambang 25 dan semuanya sudah dalam bentuk teks saat diterima collect. Prosesnya berlangsung satu nilai pada satu waktu, yaitu satu nilai mengalir melewati filter lalu map sebelum nilai berikutnya diproduksi, serupa dengan evaluasi lazy pada Sequence. Satu operator lagi yang sering dipakai adalah flowOn, yaitu operator yang mengubah dispatcher tempat builder dijalankan tanpa memengaruhi pemrosesan di hilirnya, dan pembahasannya menyambung ke topik dispatcher.

Kesalahan umum#

Kesalahan pertama adalah memperlakukan aliran dingin seakan-akan hidup terus-menerus. Builder flow dieksekusi hanya selama pengumpulan berlangsung, sehingga setelah collect selesai, tidak ada lagi kode yang berjalan, dan perubahan pada sumber data sesudahnya hanya terbaca pada pengumpulan berikutnya. Pemahaman ini penting sebelum menyimpulkan bahwa aliran mengirimkan pembaruan secara otomatis tanpa dikumpulkan.

Kesalahan kedua adalah memakai satu aliran dingin sebagai saluran kejadian yang dibagikan kepada banyak penerima. Setiap penerima yang mengumpulkan akan memicu eksekusi penuh aliran secara terpisah, sehingga sumber yang mahal, misalnya pembacaan jaringan, dipanggil berulang tanpa perlu. Kebutuhan berbagi kejadian panas antar banyak penerima ditangani jenis Flow lain pada library, sedangkan builder flow tetap menjadi pilihan untuk aliran yang memproduksi nilai per permintaan.

Ringkasan#

  • Flow adalah aliran nilai asynchronous yang dingin, dibuat dengan builder flow yang memanggil emit per nilai, dan hanya berjalan ketika operator terminal seperti collect dipanggil.
  • Setiap pengumpulan mengeksekusi builder dari awal, sehingga dua collect menghasilkan dua eksekusi penuh yang tidak berbagi keadaan.
  • Operator filter dan map dirangkai di antara sumber dan collect, dan nilainya diproses satu per satu secara lazy seperti Sequence.
  • flowOn mengubah dispatcher eksekusi builder tanpa memengaruhi pemrosesan di hilirnya.

Referensi#