Hi there. I have a question regarding Flows. How c...

# coroutines

[Doru N.]

Hi there. I have a question regarding Flows.
How come this sample code (from docs):

flow {
        // The WRONG way to change context for CPU-consuming code in flow builder
        withContext(Dispatchers.Default) {
            for (i in 1..3) {
                Thread.sleep(100) // pretend we are computing it in CPU-consuming way
                emit(i) // emit next value
            }
        }
    }
        .onEach { value -> println(value) }
        .launchIn(scope)

crashes with java.lang.IllegalStateException: Flow invariant is violated…,
but this one does not:

flowOf(1, 2, 3)
        .transformLatest {
            withContext(Dispatchers.Default) {
                Thread.sleep(100) // pretend we are computing it in CPU-consuming way
                println("emitting $it, on thread: ${Thread.currentThread().name}")
                emit(it) // emit next value
            }
        }
        .onEach {
            println("collect.onEach: $it, on thread: ${Thread.currentThread().name}")
        }
        .launchIn(scope)

to me, it should be the same case as the first (should crash with same exception). What am I missing here?
logs for 2nd sample is this:

emitting 1, on thread: DefaultDispatcher-worker-1
collect.onEach: 1, on thread: main
emitting 3, on thread: DefaultDispatcher-worker-1
collect.onEach: 3, on thread: main

[bezrukov]

transformLatest uses channelFlow under the hood which is designed to work with concurrent emitters

[bezrukov]

So you can modify your first example to use channelFlow { } builder instead of flow {} and it won't crash

[Doru N.]

is my use case poorly designed? could that be a problem that I should avoid? or is it a perfectly valid use case?

[bezrukov]

But in general you can apply .flowOn(Dispatchers.Default) to your flow to set a dispatcher for an upstream (if you just need a context switch without launching multiple concurrent coroutines)

[bezrukov]

No, seems valid to me