Jak lze přepínat vlákna ve Flow? Na kterém vlákně…

Ve výchozím nastavení Flow běží na vlákně, ze kterého byl spuštěn. Přepínání vláken lze provést pomocí operátoru flowOn.

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

// Příklad funkce, která vrací Flow
fun simpleFlow(): Flow<Int> = flow {
    logThread("Flow začal") // Logování vlákna, kde je spuštěn flow builder
    for (i in 1..3) {
        emit(i) // Emitování hodnot
    }
}

fun main() = runBlocking {
    simpleFlow()
        .flowOn(Dispatchers.IO) // Přepínáme upstream (emittery) na Dispatchers.IO
        .collect { value ->
            logThread("Sesbíráno $value") // Logování vlákna, kde se shromažďují hodnoty
        }
}

// Pomocná funkce pro logování vlákna
fun logThread(msg: String) {
    println("[$msg] [${Thread.currentThread().name}]")
}

Operátor flowOn ovlivňuje vlákno, na kterém jsou prováděny operátory před ním v řetězci (včetně flow builderu). Operátory za flowOn se provádějí na vlákně uvedeném v argumentu flowOn. Pokud je v řetězci více flowOn, každý z nich ovlivňuje část řetězce před ním.

Příklad s více flowOn:

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

fun simpleFlowWithMultipleFlowOn(): Flow<String> = flow {
    logThread("Flow builder začal") // Bude běžet na prvním specifikovaném flowOn
    emit("A")
    emit("B")
}.map {
    logThread("Mapování $it") // Bude běžet na druhém specifikovaném flowOn
    it.toLowerCase()
}.flowOn(Dispatchers.Default) // Druhý flowOn ovlivňuje map() a flow builder
.filter {
    logThread("Filtrování $it") // Bude běžet na vlákně volajícího collect (obvykle main/runBlocking)
    true
}.flowOn(Dispatchers.IO) // První flowOn ovlivňuje map() a flow builder

fun main() = runBlocking {
    simpleFlowWithMultipleFlowOn().collect { value ->
        logThread("Sbírá $value") // Běží na vlákně runBlocking (main)
    }
}

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

V tomto příkladu:

flow builder a map budou běžet na Dispatchers.Default (protože je to poslední flowOn v řetězci před nimi).
filter poběží na vlákně, ze kterého je voláno collect, tedy na vlákně, kde se provádí runBlocking (obvykle hlavní vlákno v Androidu).

Je důležité pochopit, že flowOn vytváří mezilehlý buffer a není "průhledným" přepínáním.

Jak lze přepínat vlákna ve Flow? Na kterém vlákně běží Flow?