Flow-догу жиптерди кантип которуу керек? Flow кайсы…

Өгөгдсөндө, Flow алгач иштетилген жиптен иштейт. Жиптерди которуу үчүн 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 үчүн

// Мисал функция, ал Flow кайтарат
fun simpleFlow(): Flow<Int> = flow {
    logThread("Flow башталды") // Жипти логдоо, анда flow куруучу иштетилет
    for (i in 1..3) {
        emit(i) // Маалыматтарды чыгарыш
    }
}

fun main() = runBlocking {
    simpleFlow()
        .flowOn(Dispatchers.IO) // Upstream (эмиттерлер) Dispatchers.IO кылып которулду
        .collect { value ->
            logThread("Жыйналган $value") // Жипти логдоо, анда маанилер жыйналат
        }
}

// Жипти логдоо үчүн жардамчы функция
fun logThread(msg: String) {
    println("[$msg] [${Thread.currentThread().name}]")
}

flowOn оператору иштетилген жипти таасирлейт, ал оператордун алдындагы бардык операциялар үчүн (анын ичинде flow куруучу). flowOn кийинки операциялар ошол жипте иштейт. Эгер бир нече flowOn болсо, ар бири өзүнүн алдындагы бөлүктө таасир этет.

Көп flowOn колдонулган мисал:

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

fun simpleFlowWithMultipleFlowOn(): Flow<String> = flow {
    logThread("Flow куруучу башталды") // Биринчи белгиленген `flowOn`да иштейт
    emit("A")
    emit("B")
}.map {
    logThread("Картография $it") // Экинчи белгиленген `flowOn`да иштейт
    it.toLowerCase()
}.flowOn(Dispatchers.Default) // Экинчи `flowOn` map жана flow куруучу үстүндө таасир этет
.filter {
    logThread("Фильтр $it") // `collect` чакырылган жипте иштейт (жалпысынан main/runBlocking)
    true
}.flowOn(Dispatchers.IO) // Биринчи `flowOn` map жана flow үстүндө таасир этет

fun main() = runBlocking {
    simpleFlowWithMultipleFlowOn().collect { value ->
        logThread("Жыйноо $value") // runBlocking (main) жипте иштейт
    }
}

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

Бул мисалда:

flow куруучу жана map Dispatchers.Defaultда иштейт (себеби акыркы flowOn ошол жерде)
filter collect чакырылган жипте иштейт, ал эми runBlocking иштетилген жипте (жалпысынан Main)

Мөөнөттүү мааниде, flowOn аралык буфер түзөт жана "ачык" которушуу эмес экенин түшүнүү маанилүү.

Flow-догу жиптерди кантип которуу керек? Flow кайсы жипте иштейт?