The error message is confusing here as it’s not cl...

# coroutines

[Marc Knaup]

The error message is confusing here as it’s not clear what’s happening.

I guess the code is indeed incorrect?
Then just the documentation could be improved a little.

suspend fun main() {
    flow {
        flowOf(1).collectLatest { emit(it) }
    }.collect { println(it) }
}

IllegalStateException: Flow invariant is violated

[ephemient]

you've truncated the error message, it does explain what's going on:

java.lang.IllegalStateException: Flow invariant is violated:

Emission from another coroutine is detected.

FlowCollector is not thread-safe and concurrent emissions are prohibited.

To mitigate this restriction please use 'channelFlow' builder instead of 'flow'

[ephemient]

and indeed

suspend fun main() {
    channelFlow {
        flowOf(1).collectLatest { send(it) }
    }.collect { println(it) }
}

works as expected

[Marc Knaup]

@ephemient no it doesn’t really explain the reason in this case - even if the mitigation works.
At best it presumes some knowledge about the inner workings of collectLatest.

It’s not clear that collectLatest launches multiple coroutines. And it doesn’t do anything concurrently. The explanation at the function that throws the exception only mentions launch as an example.

[ephemient]

collectLatest has to launch a coroutine so it can be cancelled on the next upstream emission

[ephemient]

the internal implementation is a little more complex because it is shared with other Flow.*Latest() methods, but it is effectively equivalent to

suspend fun <T> Flow<T>.collectLatest(action: suspend (value: T) -> Unit) {
    coroutineScope {
        fold(null) { previous: Job?, value ->
            previous?.cancelAndJoin()
            launch(start = CoroutineStart.UNDISPATCHED) {
                action(value)
            }
        }
    }
}

[Marc Knaup]

Yes that’s what I mean. One has to know quite some details before figuring out the cause of the exception.
I had that error in rather length Flow implementation. I was puzzled what could possibly cause the error since I didn’t launch any new coroutines to emit from. At least it wasn’t obvious at first.
Improving the error message or the documentation can help save developers’ time 🙂