Come si possono cambiare i thread in Flow? Su quale…

Per impostazione predefinita, Flow funziona sul thread da cui è stato avviato. È possibile cambiare thread usando l'operatore flowOn.

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

// Esempio di funzione che restituisce un Flow
fun simpleFlow(): Flow<Int> = flow {
    logThread("Flow avviato") // Log del thread in cui viene avviato il builder di flow
    for (i in 1..3) {
        emit(i) // Emetti valori
    }
}

fun main() = runBlocking {
    simpleFlow()
        .flowOn(Dispatchers.IO) // Cambia upstream (emettitori) a Dispatchers.IO
        .collect { value ->
            logThread("Raccolto $value") // Log del thread in cui vengono raccolti i valori
        }
}

// Funzione ausiliaria per il logging del thread
fun logThread(msg: String) {
    println("[$msg] [${Thread.currentThread().name}]")
}

L'operatore flowOn influisce sul thread in cui vengono eseguiti gli operatori prima di esso nella catena (incluso il flow builder). Gli operatori dopo flowOn vengono eseguiti sul thread specificato nell'argomento flowOn. Se ci sono più flowOn nella catena, ciascuno influisce sulla parte della catena prima di esso.

Esempio con più flowOn:

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

fun simpleFlowWithMultipleFlowOn(): Flow<String> = flow {
    logThread("Builder di Flow avviato") // Funziona sul primo `flowOn` specificato
    emit("A")
    emit("B")
}.map {
    logThread("Mappando $it") // Funziona sul secondo `flowOn` specificato
    it.toLowerCase()
}.flowOn(Dispatchers.Default) // Il secondo `flowOn` influisce su map() e sul builder di flow
.filter {
    logThread("Filtrando $it") // Funziona sul thread da cui viene chiamato collect (di default main/runBlocking)
    true
}.flowOn(Dispatchers.IO) // Il primo `flowOn` influisce su map() e sul builder di flow

fun main() = runBlocking {
    simpleFlowWithMultipleFlowOn().collect { value ->
        logThread("Raccolto $value") // Funziona sul thread runBlocking (main)
    }
}

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

In questo esempio:

Il flow builder e map funzionano su Dispatchers.Default (perché è l'ultimo flowOn nella catena prima di loro).
filter funziona sul thread da cui viene chiamato collect, cioè sul thread in cui viene eseguito runBlocking (di solito Main in Android).

È importante capire che flowOn crea un buffer intermedio e non è un "cambio di thread trasparente".

Come si possono cambiare i thread in Flow? Su quale thread funziona Flow?