https://kotlinlang.org logo
#coroutines
Title
# coroutines
m

Marc Knaup

08/07/2021, 10:16 PM
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.
Copy code
suspend fun main() {
    flow {
        flowOf(1).collectLatest { emit(it) }
    }.collect { println(it) }
}
IllegalStateException: Flow invariant is violated
😒 1
e

ephemient

08/08/2021, 3:01 AM
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'
1
and indeed
Copy code
suspend fun main() {
    channelFlow {
        flowOf(1).collectLatest { send(it) }
    }.collect { println(it) }
}
works as expected
m

Marc Knaup

08/08/2021, 10:22 PM
@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.
e

ephemient

08/09/2021, 1:29 AM
collectLatest has to launch a coroutine so it can be cancelled on the next upstream emission
the internal implementation is a little more complex because it is shared with other
Flow.*Latest()
methods, but it is effectively equivalent to
Copy code
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)
            }
        }
    }
}
m

Marc Knaup

08/09/2021, 4:28 AM
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 🙂
11 Views