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

Seth Madison

03/24/2023, 9:41 PM
I think I am misunderstanding the semantics of
transform
. I have a flow along the lines of`sharedFlow -> transformWhile -> map` , and I see the
map
run even when the downstream collector has completed. Is this expected? Example in the thread.
Copy code
fun testFlo() {
    val coroutineScope = CoroutineScope(Dispatchers.Default)
    val sharedFlo = MutableSharedFlow<Int>(0, Int.MAX_VALUE, BufferOverflow.SUSPEND)

    coroutineScope.launch {
        var i = 0
        repeat(100) {
            sharedFlo.emit(i)
            i +=1
            delay(10)
        }
    }
	
    var mapCount = 0
    val mappedFlo = sharedFlo.transformWhile { 
        emit(it)
        true
    }.map {
        mapCount += 1
        if (mapCount > 2) throw RuntimeException("FAIL")
        it
    }.flowOn(<http://Dispatchers.IO|Dispatchers.IO>)

    runBlocking {
        mappedFlo
            .take(2)
            .catch { println("caught error $it") }
            .collect {}
    }
}
In this example, the catch will get triggered every so often. Worse yet, I think there are cases where the exception is thrown and the catch doesn’t kick in.
e

ephemient

03/24/2023, 9:50 PM
there is a difference in behavior between cold streams (values produced when demanded) and hot streams (values pushed to consumers)
this simplified example will consistently trigger "hot" but not "cold": https://pl.kotl.in/BmqYuv-VK
s

Seth Madison

03/24/2023, 10:27 PM
Right, I guess I just would’ve assumed that even for a hot flow, the value would never end up making it to the map. Like, I would assume that once the subscribing flow is cancelled, everything upstream from it UP TO the hot flow would get cancelled atomically.
e

ephemient

03/24/2023, 10:29 PM
in the hot case (and also to a limited extent in
flowOn
due to the channel buffer) the upstream emits asynchronously to the downstream collector
s

Seth Madison

03/24/2023, 10:34 PM
Got it. And I suppose there’s no way to know if the downstream is still subscribed? To take a simple example:
Copy code
hotFlow.map { if (it.someVal) throw RuntimeException() }.collect { }
Is there any way to prevent that exception from getting thrown the collect has completed?
Because right now, short of a CoroutineExceptionHandler on the scope of the hotFlow I can’t see a way to prevent that from crashing my app.
Right?
e

ephemient

03/24/2023, 10:36 PM
you can get some feedback with
SharingStarted
but it's asynchronous so you can't use it to control this
s

Seth Madison

03/24/2023, 10:39 PM
The CEH would need to be on the scope of the collect, right? Or would it be on the scope of the hotflow? Intuitively I would think it would be on the collect, since that’s what is running the map, but since the collect already completed, I’m confused.
e

ephemient

03/24/2023, 10:40 PM
CEH doesn't do anything in most scopes because they propagate cancellation
GlobalScope and SupervisorScope don't propagate cancellation so you could add the handler to them
(in which case your app will not crash regardless of whether you install a handler or not)
s

Seth Madison

03/24/2023, 10:47 PM
Got it. Makes sense.
erm, actually..
I guess the thing I’m confused about is which scope the crash ends up propagating to.
Like:
Copy code
scope a = CoroutineScope()
scope b = CoroutineScope()

hotFlow = flow {
}.shareIn(a)

b.launch {
  hotFlow.map { if (it.someVal) throw RuntimeException() }.collect {}
}
In that example, if the exception triggers after the collect completes, does the exception go to
a
or
b
?
e

ephemient

03/24/2023, 11:19 PM
b
s

Seth Madison

03/24/2023, 11:21 PM
I guess that makes sense
4 Views