Kaip galima perjungti srautus Flow? Kurioje srauto…

Pagal numatytuosius nustatymus Flow veikia tame pačia srautu, iš kurio jis buvo paleistas. Srautų perjungimą galima atlikti naudojant operatorių flowOn.

import kotlinx.coroutines.CoroutineDispatcher // CoroutineDispatcher
import kotlinx.coroutines.Dispatchers // Dispatchers
import kotlinx.coroutines.flow.Flow // Flow
import kotlinx.coroutines.flow.flow // flow
import kotlinx.coroutines.flow.flowOn // flowOn
import kotlinx.coroutines.runBlocking // runBlocking

// Pavyzdinė funkcija, grąžinanti Flow
fun simpleFlow(): Flow<Int> = flow {
    logThread("Flow pradėtas") // Srauto logavimas, kur vykdomas flow builder
    for (i in 1..3) {
        emit(i) // Reikšmių išsiuntimas
    }
}

fun main() = runBlocking {
    simpleFlow()
        .flowOn(Dispatchers.IO) // Perjungiame upstream (emiterius) į Dispatchers.IO
        .collect { value ->
            logThread("Surinkta $value") // Srauto logavimas, kur renkamos reikšmės
        }
}

// Pagalbinė funkcija srauto logavimui
fun logThread(msg: String) {
    println("[$msg] [${Thread.currentThread().name}]")
}

Operatorius flowOn veikia srauto, kuriame vykdomi operatoriai prieš jį, kontekste (įskaitant flow builderį). Operatoriai po flowOn vykdomi nurodytu srautu, kuris yra nurodytas flowOn argumentu. Jei yra keli flowOn, kiekvienas jų veikia savo dalį prieš jį.

Pavyzdys su keliais flowOn:

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun simpleFlowWithMultipleFlowOn(): Flow<String> = flow {
    logThread("Flow kūrėjas pradėtas") // Veiks pirmame nurodytame `flowOn`
    emit("A")
    emit("B")
}.map {
    logThread("Mapinimas $it") // Veiks antrame nurodytame `flowOn`
    it.toLowerCase()
}.flowOn(Dispatchers.Default) // Antras `flowOn` veikia ant `map` ir `flow` kūrėjo
.filter {
    logThread("Filtravimas $it") // Veiks ant srauto, iš kurio kviečiama `collect` (paprastai main/runBlocking)
    true
}.flowOn(Dispatchers.IO) // Pirmas `flowOn` veikia ant `map` ir `flow` kūrėjo

fun main() = runBlocking {
    simpleFlowWithMultipleFlowOn().collect { value ->
        logThread("Rinkimas $value") // Veikia `runBlocking` (main) sraute
    }
}

fun logThread(msg: String) {
    println("[$msg] [${Thread.currentThread().name}]")
}

Šiame pavyzdyje:

flow kūrėjas ir map veiks Dispatchers.Default (nes tai yra paskutinis flowOn jų prieš)
filter veiks sraute, iš kurio kviečiama collect, tai yra runBlocking (paprastai pagrindinis srautas Android'e)

Svarbu suprasti, kad flowOn sukuria tarpinį buferį ir nėra "permatomas" perjungimas.

Kaip galima perjungti srautus Flow? Kurioje srauto dalyje veikia Flow?