Flow — ano ito, cold at hot stream sa Kotlin coroutines

Ano ang Flow sa Kotlin?

Flow — ay isang uri mula sa paketeng kotlinx.coroutines.flow, na kumakatawan sa isang malamig na asynchronous na daloy ng data. Sa esensya, ang Flow ay isang coroutine sequence na naglalabas ng mga halaga sa pamamagitan ng function na emit() at nagtatapos alinman sa tagumpay o sa isang exception. Ang koleksyon ng daloy ay ginagawa sa pamamagitan ng terminal operator na collect(), na isang suspend-function.

Malamig na Semantiko

Cold stream ay nangangahulugang ang kodigo sa loob ng flow-builder ay isinasagawa muli para sa bawat subscriber. Ang Observable.fromIterable sa RxJava ay kumikilos nang katulad: ang isang bagong subscriber ay tumatanggap ng lahat ng halaga mula sa simula. Sa Flow, ito ay ipinatupad sa pamamagitan ng suspend-function na collect, na bumabara sa coroutine sa buong tagal ng pagtitipon ng data.

Flow builders

Ang Kotlin ay nagbibigay ng ilang paraan upang lumikha ng Flow: flow { } — pangunahing konstruksyon na may emit(), flowOf(vararg values) — para sa isang nakapirming hanay ng mga halaga, .asFlow() — extension para sa mga koleksyon at Sequence. Lahat ng builder ay malamig — ang data ay nabuo lamang kapag ang terminal operator ay tinawag.

Cold at Hot stream

Ang paghahati sa cold at hot stream ay isang pangunahing konsepto ng reaktibong programming. Cold stream (Flow, Observable) ay nagsisimula ng pagbuo ng data sa pag-subscribe. Ang Hot stream (Channel, SharedFlow) ay naglalabas ng data nang hiwalay — ang subscriber ay tumatanggap lamang ng kung ano ang nangyayari pagkatapos ng subscription, nang walang simula ng sequence.

SharedFlow — ay isang mainit na Flow na maaaring magkaroon ng maraming subscriber at maaaring mag-replay ng mga huling halaga kapag nakatakda ang replay. Ang SharedFlow ay angkop para sa mga kaganapan (isang beses na notification). StateFlow — ang variant nito na may nakapirming halaga ng estado, na nag-cache ng huling halaga para sa mga bagong subscriber.

ChannelFlow ay gumagamit ng Channel sa ilalim, na pinagsasama ang mga katangian ng Flow at Channel. Sinusuportahan nito ang buffering at backpressure sa pamamagitan ng kapasidad (capacity). Ang ChannelFlow ay kapaki-pakinabang sa pag-convert ng callback-API sa isang reaktibong daloy, kapag ang mga halaga ay inilalabas mula sa iba't ibang coroutine.

Pag-convert sa pagitan ng cold at hot

Para sa pag-convert ng cold Flow sa hot SharedFlow ay ginagamit ang operator na shareIn(scope, started, replay). Kinokontrol ng parameter na started ang sandali ng pagsisimula: SharingStarted.WhileSubscribed() — aktibo habang may mga subscriber, Lazily — simula sa unang subscriber, Eagerly — agarang simula. Baliktad na conversion — hot sa cold: Ang StateFlow.asFlow() ay nagbabalik ng malamig na Flow na sa collect ay naglalabas ng kasalukuyang halaga ng StateFlow. Ito ay maginhawa para sa pagsubok.

Mga Operator ng Flow

Flow ay nagbibigay ng mayamang hanay ng mga operator na gumagana bilang suspend-function sa loob ng coroutine. Ang mga operator ay walang estado at nagbabalik ng bagong Flow — ang orihinal na daloy ay nananatiling hindi nagbabago. Ito ay nagpapahintulot sa pagbuo ng mga ligtas na kadena ng pagbabago nang walang mga epekto.

Ang operator na map ay nagbabago ng bawat halaga ng daloy sa pamamagitan ng isang asynchronous o synchronous na pagbabago. filter ay nagpapadaan lamang ng mga halaga na nakakatugon sa kondisyon. catch ay humahawak ng mga exception bago ang terminal operator at nagpapahintulot sa pagpapanumbalik ng daloy. flatMapLatest ay kinakansela ang nakaraang paglabas kapag dumating ang bagong halaga — katulad ng switchMap sa Rx.

Ang operator na debounce sa Flow ay naantala ang publikasyon ng isang halaga ng isang tinukoy na timeout. Kung sa panahong ito ay dumating ang isang bagong halaga — ang timer ay nire-reset. Sa Android, ang debounce ay ginagamit para sa paghahanap: ang kahilingan ay ipinapadala lamang pagkatapos ng isang pause na 300-400 ms, na binabawasan ang bilang ng mga tawag sa API ng 3-5 beses.

Mga Terminal na Operator

Bukod sa collect(), ang Flow ay sumusuporta sa iba pang terminal operator: toList() ay nagtitipon ng lahat ng halaga sa isang listahan — kapaki-pakinabang para sa mga pagsubok, first() ay nagbabalik ng unang elemento at kinakansela ang daloy, single() ay umaasa ng eksaktong isang elemento. fold(initial) ay nag-iipon ng mga halaga sa pamamagitan ng ibinigay na function. Lahat ng terminal operator ay suspend-function at dapat tawagin sa loob ng isang coroutine o ibang suspend-function.

Mga Halimbawa ng Kodigo ng Flow

Unang halimbawa — pangunahing Flow na may pagbuo ng mga numero at pagbabago sa pamamagitan ng operator na map:

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

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

Ikalawang halimbawa — pagbabago ng daloy na may pagsala at pangangasiwa ng error sa pamamagitan ng catch:

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

Ikatlong halimbawa — paggamit ng StateFlow sa ViewModel para sa reaktibong UI sa 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 at SharedFlow

StateFlow — ay isang mainit na Flow na may iisang kasalukuyang halaga. Naka-cache nito ang huling halaga at agad itong ipinapasa sa bagong subscriber. Ang StateFlow ay isang Observable na lalagyan para sa estado, sumusuporta sa equals na paghahambing — kung ang bagong halaga ay tumutugma sa kasalukuyan, ang paglabas ay hindi nagaganap. Ang Jetpack Compose ay gumagamit ng StateFlow sa pamamagitan ng collectAsState().

SharedFlow — ay isang mas nababaluktot na mainit na Flow nang walang sapilitang paunang halaga. Ang SharedFlow ay naka-configure sa pamamagitan ng replay (bilang ng mga halaga para sa mga bagong subscriber), extraBufferCapacity (buffer sa labas ng replay), at onBufferOverflow (estratehiya sa pag-apaw). Ang SharedFlow ay perpekto para sa isang beses na mga kaganapan: nabigasyon, Snackbar, analitika.

Flow sa arkitektura ng Android ay inirerekomenda ng Google bilang pangunahing mapagkukunan ng data (Layer: Repository → UseCase → ViewModel). Ang LiveData ay mas mababa sa Flow sa flexibility: Sinusuportahan ng Flow ang mga coroutine, operator, backpressure at gumagana sa labas ng layer ng UI. Ang paglipat mula LiveData patungong Flow ay karaniwang kasanayan sa mga modernong proyekto ng Android.

Kapag gumagamit ng Flow sa ViewModel, mahalaga ang tamang pagpili ng uri. StateFlow ay perpekto para sa estado ng UI na dapat makaligtas sa pag-ikot ng screen. Ang SharedFlow ay angkop para sa mga kaganapan kung saan ang muling pagproseso ay hindi katanggap-tanggap — halimbawa, nabigasyon. Ang Flow na may collect() sa lifecycleScope ay nagbibigay ng maximum na kontrol sa konteksto ng pagpapatupad, ngunit nangangailangan ng manu-manong pagkansela kapag lumalabas sa screen.

Ang pagsubok ng Flow ay ginagawa sa pamamagitan ng kotlinx-coroutines-test. Ang library ay nagbibigay ng TestDispatcher — virtual na oras na nagpapahintulot sa pagpapabilis ng mga pagkaantala (delay) at pagkontrol sa pagkakasunud-sunod ng pagpapatupad ng mga coroutine. Ang TestScope.runTest { } ay lumilikha ng isang nakahiwalay na kapaligiran para sa pagsubok ng Flow. Ang operator na toList() ay madalas na ginagamit sa mga pagsubok upang tipunin ang lahat ng halaga ng flow na may timeout, upang suriin kung ang daloy ay naglabas ng tamang pagkakasunod-sunod ng data.

Ang Flow ay mahusay na nagsasama sa Room (library ng Android para sa mga database): ang mga pamamaraan ng DAO ay maaaring magbalik ng Flow<List<Entity>>. Awtomatikong naglalabas ang Room ng bagong halaga sa anumang pagbabago sa talahanayan — ang UI ay naa-update nang walang manu-manong trigger. Ito ay ipinatupad sa pamamagitan ng InvalidationTracker, na sa ilalim ay gumagamit ng Flow na may callbackFlow. Ang ganitong paraan ay nag-aalis ng pangangailangan para sa LiveData at ginagawang ganap na coroutine-oriented ang layer ng data. Ang Jetpack Compose sa pamamagitan ng collectAsState() ay nag-subscribe sa StateFlow at muling iginuguhit lamang ang mga bahaging iyon na ang data ay nagbago — ito ay nagbibigay ng pagganap na hindi maaabot sa mga arkitekturang nakatuon sa LiveData. Ang DataStore (kapalit ng SharedPreferences) ay nagbabalik din ng Flow<Preferences>, na tinitiyak ang reaktibong pagbabasa ng mga setting ng app nang walang manu-manong mga trigger ng pag-update.

Sinusuportahan ng Flow ang interprocess na komunikasyon sa pamamagitan ng kotlinx-coroutines-core sa JVM nang walang karagdagang library. Halimbawa, sa mga server application sa Ktor, ang Flow ay maaaring kumatawan sa isang daloy ng mga papasok na WebSocket na mensahe. Ang bawat mensahe ay inilalabas sa daloy, dumadaan sa pagsala at pagsasama-sama sa pamamagitan ng mga operator, at ang resulta ay ipinapadala sa kliyente. Ang ganitong paraan ay pumapalit sa mga reaktibong library tulad ng Reactor o RxJava sa mga proyekto ng Kotlin.

Ang pagiging tugma ng Flow sa umiiral na RxJava code ay ibinibigay ng modyul na kotlinx-coroutines-rx3. Ang extension function na Flow.asObservable() ay nagko-convert ng Flow patungong Observable mula sa RxJava 3. Baliktad na conversion — CompletableSource.asFlow(), Observable.asFlow(). Pinapasimple nito ang paglipat mula RxJava patungong coroutine: ang proyekto ay maaaring isulat muli nang paunti-unti, na iniiwan ang bahagi ng mga layer sa RxJava. Sa conversion, kailangang isaalang-alang ang pagkakaiba sa cold/hot semantika: ang Observable ay maaaring parehong cold at hot, ang Flow ay palaging cold para sa ordinaryong Flow at hot para sa SharedFlow.

Pangangasiwa ng Error at Pagsubok ng Flow

Para sa pangangasiwa ng error sa Flow ay may kakaiba: kung ang exception ay nangyari sa loob ng flow-builder bago ang terminal operator, ito ay ipinapasa sa catch. Kung ang exception ay nangyari sa operator pagkatapos ng builder, ito ay hinahawakan ng catch pagkatapos ng operator na ito. retryWhen ay nagpapahintulot sa pag-ulit ng subscription na may kondisyon: ulitin sa network error hanggang 3 beses, ngunit huwag ulitin sa CancellationException. Inaalis ng Flow ang mga error na nakadepende sa estado dahil hindi ito nag-iimbak ng estado — pinapasimple nito ang debugging kumpara sa Observable, kung saan ang Subject ay nag-iimbak ng panloob na estado.

Ang pagsubok ng Flow na may kotlinx-coroutines-test ay gumagamit ng TestDispatcher para sa simulation ng mga pagkaantala. Ang Turbine — sikat na library mula sa komunidad para sa pagsubok ng Flow: test { } ay nagpapatakbo ng Flow, ang awaitItem() ay naghihintay ng susunod na halaga, ang awaitComplete() ay naghihintay ng pagkumpleto. Ang Turbine ay nagdaragdag ng default na timeout, na pumipigil sa pagbitin ng mga pagsubok. Para sa pagsubok ng StateFlow, gamitin ang .testIn(scope) na may pagsusuri ng mga halaga sa kronolohikal na pagkakasunud-sunod.