Ինչպե՞ս կարելի է փոխել հոսքերը 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 builder-ը
    for (i in 1..3) {
        emit(i) // Վերադարձնում արժեքներ
    }
}

fun main() = runBlocking {
    simpleFlow()
        .flowOn(Dispatchers.IO) // Փոխում ենք upstream (emit-ները) Dispatchers.IO-ին
        .collect { value ->
            logThread("Ստացված $value") // Լոգավորում հոսքի, որտեղ հավաքվում են արժեքները
        }
}

// Օգնական ֆունկցիա հոսքի լոգավորման համար
fun logThread(msg: String) {
    println("[$msg] [${Thread.currentThread().name}]")
}

flowOn օպերատորը ազդում է այն հոսքի վրա, որտեղից սկսվում են գործողությունները առաջ նրանից (ներառյալ flow builder-ը): Օպերատորները հետո flowOn աշխատում են նշված հոսքում, որը փոխանցվում է flowOn-ի արգումենտով։ Եթե մի քանի flowOn-ներ կան, յուրաքանչյուրն ազդում է իր նախորդ հատվածի վրա։

Նմուշ՝ մի քանի flowOn-ներով:

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

fun simpleFlowWithMultipleFlowOn(): Flow<String> = flow {
    logThread("Flow builder սկսվեց") // Աշխատում է առաջին նշված `flowOn`-ով
    emit("A")
    emit("B")
}.map {
    logThread("Mapping $it") // Աշխատում է երկրորդ նշված `flowOn`-ով
    it.toLowerCase()
}.flowOn(Dispatchers.Default) // Երկրորդ `flowOn`-ը ազդում է `map`-ի և `flow`-ի վրա
.filter {
    logThread("Filtering $it") // Աշխատում է `collect`-ից դուրս հոսքի վրա (համարապես main/runBlocking)
    true
}.flowOn(Dispatchers.IO) // Առաջին `flowOn`-ը ազդում է `map`-ի և `flow`-ի վրա

fun main() = runBlocking {
    simpleFlowWithMultipleFlowOn().collect { value ->
        logThread("Collecting $value") // Աշխատում է `runBlocking`-ի հոսքում (main)
    }
}

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

Այս օրինակով՝

• flow builder և map աշխատում են Dispatchers.Default-ում (քանի որ դա վերջին flowOn-ն է շարքում)
• filter աշխատում է այն հոսքում, որտեղից collect-ն է կանչվում, այսինքն՝ runBlocking-ի հոսքում (հատկապես Android-ում main հոսքում)

Կարևոր է հասկանալ, որ flowOn ստեղծում է միջանկյալ բուֆեր և չի հանդիսանում "փայլուն" փոխարկում։