SharedFlow

Pada chapter Shared Flows kita sudah melihat SharedFlow sebagai multicast event stream. Sekarang kita mendalami MutableSharedFlow sebagai event bus — bagaimana mengonfigurasi buffer overflow, kapan memakai tryEmit versus emit, dan pola untuk membungkus event bus di class. Bab ini fokus pada konfigurasi dan idiom produksi yang sering dijumpai di aplikasi real-time.

Konfigurasi MutableSharedFlow #

Tiga parameter utama saat membuat MutableSharedFlow:

Tiga strategi BufferOverflow:

Total kapasitas buffer = replay + extraBufferCapacity. Buffer penuh saat semua slot terisi dan belum ada collector yang memproses.

EventBus dengan SharedFlow #

Kode berikut mendemonstrasikan event bus sederhana yang membungkus MutableSharedFlow dengan dua API publikasi: tryEmit (non-suspend) dan emit (suspend):

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.*

// Event bus sederhana memakai SharedFlow
class EventBus {
    private val _events = MutableSharedFlow<String>(
        replay = 0,
        extraBufferCapacity = 4,
        onBufferOverflow = BufferOverflow.DROP_OLDEST,
    )
    val events: SharedFlow<String> = _events.asSharedFlow()

    fun publish(event: String) {
        // tryEmit: non-suspend, kembalikan Boolean
        val ok = _events.tryEmit(event)
        println("[bus] tryEmit('$event') -> $ok")
    }

    suspend fun publishSuspend(event: String) {
        _events.emit(event)
        println("[bus] emit('$event') sukses")
    }
}

fun main() = runBlocking {
    val bus = EventBus()

    // Subscribe sebelum publish
    val job = launch {
        bus.events.collect { println("[subscriber] terima: $it") }
    }
    delay(50)

    // tryEmit: tidak menangguhkan, langsung kembali
    bus.publish("event-1")
    bus.publish("event-2")
    delay(30)

    // emit: suspend function
    bus.publishSuspend("event-3")
    delay(30)

    // Tidak ada replay: subscriber terlambat tidak menerima event lama
    println()
    println("subscriber terlambat mulai:")
    val jobTerlambat = launch {
        bus.events.take(1).collect { println("[terlambat] terima: $it") }
    }
    delay(20)
    bus.publish("event-4")
    delay(30)
    job.cancel()
    jobTerlambat.cancel()
}

Output yang Dihasilkan:

[bus] tryEmit('event-1') -> true
[bus] tryEmit('event-2') -> true
[subscriber] terima: event-1
[subscriber] terima: event-2
[bus] emit('event-3') sukses
[subscriber] terima: event-3

subscriber terlambat mulai:
[bus] tryEmit('event-4') -> true
[subscriber] terima: event-4
[terlambat] terima: event-4

Mari kita telaah baris-baris kode di atas:

1. Baris 8-12 (MutableSharedFlow<String>(replay = 0, extraBufferCapacity = 4, onBufferOverflow = DROP_OLDEST)): konfigurasi event bus murni. replay=0 berarti event yang sudah dipublikasikan tidak disimpan — subscriber terlambat tidak menerima event lama. extraBufferCapacity=4 memberi ruang untuk 4 event yang belum diproses subscriber. DROP_OLDEST memastikan tryEmit tidak pernah gagal — nilai tertua dibuang saat buffer penuh.
2. Baris 14 (val events: SharedFlow<String> = _events.asSharedFlow()): mengekspos versi read-only. Pemanggil hanya bisa collect, tidak bisa tryEmit/emit dari luar. Pola ini melindungi integritas bus — hanya class EventBus yang boleh mempublikasikan.
3. Baris 17-21 (fun publish(event: String)): API non-suspend memakai tryEmit. Mengembalikan Boolean — true jika berhasil mengirim (atau masuk buffer), false jika buffer penuh dan onBufferOverflow adalah SUSPEND. Karena di sini strategi adalah DROP_OLDEST, tryEmit selalu mengembalikan true.
4. Baris 23-26 (suspend fun publishSuspend): API suspend memakai emit. Menangguhkan jika buffer penuh dan strategi SUSPEND. Cocok dipanggil dari dalam coroutine — memiliki backpressure alami.
5. Baris 33-35 (launch { bus.events.collect { ... } }): subscriber mulai. Karena delay(50) setelahnya, subscriber pasti sudah siap sebelum event dipublikasikan.
6. Baris 39-41 (bus.publish("event-1"), dst.): dua event dipublikasikan dengan tryEmit. Output menunjukkan tryEmit selalu berhasil (-> true). Subscriber menerima kedua event.
7. Baris 49-51 (subscriber terlambat): subscriber baru mulai setelah event 1-3 dipublikasikan. Karena replay=0, subscriber terlambat hanya menerima event yang dipublikasikan SETELAH ia mulai — yaitu event-4. Output [terlambat] terima: event-4 membuktikan hal ini.

tryEmit vs emit #

Dua API publikasi memiliki use case berbeda:

button.onClick { event ->
    bus.tryEmit(event)  // tidak boleh suspend
}

launch {
    for (item in data) {
        bus.emit(item)  // menangguhkan jika collector lambat
    }
}

Pola EventBus di Aplikasi Nyata #

Event bus umum dipakai untuk: navigasi, notifikasi, event UI lintas komponen, atau status jaringan. Pola idiomatik:

object AppEvents {
    private val _bus = MutableSharedFlow<AppEvent>(
        extraBufferCapacity = 16,
        onBufferOverflow = BufferOverflow.DROP_OLDEST,
    )
    val bus: SharedFlow<AppEvent> = _bus.asSharedFlow()

    fun emit(event: AppEvent) = _bus.tryEmit(event)
}

sealed interface AppEvent {
    data class Login(val user: String) : AppEvent
    data class Logout(val reason: String) : AppEvent
    object NetworkRestored : AppEvent
}

Pemakaian:

// di ViewModel
viewModelScope.launch {
    AppEvents.bus.collect { event ->
        when (event) {
            is AppEvent.Login -> updateUser(event.user)
            is AppEvent.Logout -> clearUser()
            AppEvent.NetworkRestored -> refreshAll()
        }
    }
}

// di tempat lain
AppEvents.emit(AppEvent.Login("budi"))

Referensi Lanjutan #

• Flow — panduan resmi SharedFlow untuk event broadcast
• kotlinx.coroutines core — API reference kotlinx.coroutines.flow.SharedFlow

Selanjutnya #

SharedFlow adalah stream hot — aktif terlepas dari collector. Untuk stream cold — yang baru menghasilkan nilai saat di-collect — Kotlin menyediakan Flow<T>. Bab Flow Fundamentals akan memperkenalkan builder flow { emit(...) }, semantik lazy, dan perbedaan mendasar dengan SharedFlow.