How can you switch threads in Flow? On which thread…

By default, Flow operates on the thread it was launched on. You can switch threads using the flowOn operator.

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

// Example of a function returning a Flow
fun simpleFlow(): Flow<Int> = flow {
    logThread("Flow started") // Log the thread where flow builder runs
    for (i in 1..3) {
        emit(i) // Emit values
    }
}

fun main() = runBlocking {
    simpleFlow()
        .flowOn(Dispatchers.IO) // Switch upstream (emitters) to Dispatchers.IO
        .collect { value ->
            logThread("Collected $value") // Log the thread where values are collected
        }
}

// Helper function for logging thread info
fun logThread(msg: String) {
    println("[$msg] [${Thread.currentThread().name}]")
}

The flowOn operator affects the thread where upstream operators (including the flow builder) run. Operators after flowOn run on the thread specified in its argument. Multiple flowOn operators affect different parts of the chain.

Example with multiple flowOn:

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

fun simpleFlowWithMultipleFlowOn(): Flow<String> = flow {
    logThread("Flow builder started") // Will run on the last specified flowOn before it
    emit("A")
    emit("B")
}.map {
    logThread("Mapping $it") // Will run on the second specified flowOn
    it.toLowerCase()
}.flowOn(Dispatchers.Default) // Second flowOn affects map() and flow builder
.filter {
    logThread("Filtering $it") // Will run on the thread where collect() is called (default main/runBlocking)
    true
}.flowOn(Dispatchers.IO) // First flowOn affects map() and flow builder

fun main() = runBlocking {
    simpleFlowWithMultipleFlowOn().collect { value ->
        logThread("Collecting $value") // Runs on runBlocking thread (main)
    }
}

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

In this example:

flow builder and map run on Dispatchers.Default (the last flowOn before them in the chain).
filter runs on the thread where collect is called (usually main thread).

Understanding flowOn is crucial for thread management in Kotlin Flow pipelines. It creates an intermediate buffer and does not act as a transparent thread switch.

How can you switch threads in Flow? On which thread does Flow operate?