Shared Flows

Pada chapter sebelumnya, kita memakai Channel untuk komunikasi satu-ke-satu antar coroutine. Bagaimana jika satu sumber event ingin memberi tahu banyak subscriber sekaligus? Pola ini umum di aplikasi nyata: event UI, notifikasi real-time, atau update harga saham. SharedFlow adalah jawaban Kotlin — hot multicast stream di mana semua collector aktif menerima nilai yang sama. Bab ini mengenalkan konsep SharedFlow, perbedaan dengan Channel, dan dua konfigurasi utama: replay dan buffer overflow.

Apa itu SharedFlow #

SharedFlow<T> adalah sebuah hot stream — ia aktif terlepas dari ada atau tidaknya collector. Saat sebuah nilai di-emit, semua collector yang sedang aktif menerima nilai tersebut. Karakteristik utama:

Perbedaan dengan Channel:

Demonstrasi Multicast #

Kode berikut mendemonstrasikan dua fitur SharedFlow: multicast dan replay buffer:

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

fun main() = runBlocking {
    // 1. SharedFlow multicast: banyak collector menerima emit yang sama
    println("=== 1. SharedFlow multicast ===")
    val shared = MutableSharedFlow<String>(replay = 0)

    // Collector A mulai
    val jobA = launch {
        shared.collect { println("[A] terima: $it") }
    }
    // Collector B mulai
    val jobB = launch {
        shared.collect { println("[B] terima: $it") }
    }

    delay(50)  // pastikan collector siap
    shared.emit("event-1")
    shared.emit("event-2")

    delay(50)
    jobA.cancel()
    jobB.cancel()

    // 2. SharedFlow dengan replay buffer
    println()
    println("=== 2. SharedFlow dengan replay=2 ===")
    val denganReplay = MutableSharedFlow<Int>(replay = 2)
    denganReplay.emit(10)
    denganReplay.emit(20)
    denganReplay.emit(30)  // nilai 10 diturunkan dari buffer replay

    // Collector terlambat menerima replay buffer
    println("collector terlambat menerima:")
    denganReplay.take(2).collect { println("  - $it") }
}

Output yang Dihasilkan:

=== 1. SharedFlow multicast ===
[A] terima: event-1
[B] terima: event-1
[A] terima: event-2
[B] terima: event-2

=== 2. SharedFlow dengan replay=2 ===
collector terlambat menerima:
  - 20
  - 30

Mari kita telaah baris-baris kode di atas:

1. Baris 7 (MutableSharedFlow<String>(replay = 0)): membuat SharedFlow dengan replay=0. Nilai yang di-emit saat tidak ada collector hilang. replay=0 cocok untuk event broker murni — hanya collector aktif yang menerima.
2. Baris 10-15 (dua collector A dan B): masing-masing launch sebuah coroutine yang memanggil shared.collect { ... }. collect adalah suspend function yang menangguhkan coroutine sampai nilai tersedia. Karena SharedFlow multicast, kedua collector menerima setiap emit.
3. Baris 18-19 (emit("event-1"), emit("event-2")): mengirim dua nilai. Karena replay=0 dan kedua collector sudah aktif (setelah delay(50)), kedua nilai diterima oleh A dan B. Output menunjukkan keduanya mencetak “event-1” lalu “event-2”.
4. Baris 28 (MutableSharedFlow<Int>(replay = 2)): membuat SharedFlow dengan replay=2. Dua nilai terakhir yang di-emit akan diingat dan dikirim ke collector baru saat mereka mulai mengumpulkan.
5. Baris 29-31 (emit(10), emit(20), emit(30)): mengirim tiga nilai. Karena replay=2, hanya dua nilai terakhir yang disimpan di buffer replay. Nilai 10 tidak lagi tersimpan setelah 30 di-emit.
6. Baris 35 (denganReplay.take(2).collect { ... }): collector baru memulai. Karena replay=2, collector menerima nilai 20 dan 30 (dua nilai terakhir) — bukan 10. Operator take(2) membatasi collect ke dua nilai pertama.

Konfigurasi SharedFlow #

Tiga parameter utama saat membuat MutableSharedFlow:

Untuk emit() (suspend), SharedFlow dapat menangguhkan jika buffer penuh dan strategi adalah SUSPEND. Untuk tryEmit() (non-suspend), SharedFlow akan mengembalikan false jika tidak bisa mengirim — tidak menangguhkan.

val flow = MutableSharedFlow<String>(
    replay = 1,
    extraBufferCapacity = 4,
    onBufferOverflow = BufferOverflow.DROP_OLDEST,
)

Konfigurasi di atas: replay 1 nilai, buffer tambahan 4, dan drop nilai tertua saat buffer penuh — tidak pernah menangguhkan producer.

Hot vs Cold #

SharedFlow adalah hot — ia memancarkan nilai terlepas dari ada atau tidaknya collector. Kontras dengan Flow (cold) yang baru mulai menghasilkan nilai saat di-collect. Perbedaan ini memiliki implikasi praktis:

Referensi Lanjutan #

• Flow — panduan resmi SharedFlow sebagai hot multicast stream
• kotlinx.coroutines core — API reference kotlinx.coroutines.flow

Selanjutnya #

Setelah memahami konsep SharedFlow sebagai multicast event broker, bab StateFlow akan memperkenalkan varian khusus: state holder yang selalu memiliki nilai, dengan semantik conflation (hanya nilai terbaru yang dikirim ke collector baru).