Hello guys, Is it ok to do something like that: ...

# coroutines

[Radoslaw Juszczyk]

Hello guys,

Is it ok to do something like that:

someFlow() // emits [a, b, c, ... ]
    .onStart {
        CoroutineScope(coroutineContext).launch {
            var c = 0
            //do stuff
            while(true) {
                delay(500)
                c++
                Timber.d("c = $c")
            }
        }
    }

I want to have a flow with a side effect, so something is done independently while the flow is collected but it is stopped when the flow is no longer collected

so I want to have logs like that:

val job = viewModelScope.launch {
    someFlow().collectLatest {  Timber.d("someFlow $it") }
}

//now logs should be:
c = 1
c = 2
c = 3
someFlow a
c = 4
c = 5
someFlow b
c = 6
someFlow c
c = 7

job.cancel()

//logs stop and I do not get c = 8, c = 9 anymore (internal job canceled)

[Sam]

Creating the custom CoroutineScope isn’t a great solution. The flow doesn’t have its own Job, so your new scope becomes a child of whatever job collects the flow. That works in your example because you’re creating a new job specifically to collect that flow, but a flow can just as easily be collected inline as part of an existing coroutine. At that point it would become very hard to cancel or keep track of the side-effect job.

[Sam]

Better to use the coroutineScope builder function to make an inner scope that is tied to the lifecycle of the flow. That way it’s fully encapsulated and the coroutines it creates can never leak outside the flow. Here’s my approach: https://pl.kotl.in/e0yf0JDe_

[Sam]

fun <T> Flow<T>.withSideEffect() = flow {
    coroutineScope {
        val sideEffect = launch {
            var c = 0
            //do stuff
            while (true) {
                delay(500)
                c++
                println("c = $c")
            }
        }
        onCompletion {
            sideEffect.cancel()
        }.collect {
            emit(it)
        }
    }
}

[Radoslaw Juszczyk]

sweet, will it cancel the sideEffect if I 'cancel' the calling coroutine before the flow completes?

[Sam]

Yep 👍 there are two things working to control the lifecycle:
1. The coroutineScope is encapsulated inside the flow so any coroutines it launches will always terminate when the downstream flow collector stops collecting the flow. That includes if the collector coroutine is cancelled.
2. The onCompletion block explicitly cancels the sideEffect job any time the upstream flow terminates (normally or exceptionally)

[Radoslaw Juszczyk]

Thank you so much, just gave it a test and it works perfectly.

In our case both worked the same way because our Flow was actually a StateFlow so it never completes.

Btw: is there any difference between using corotineScope vs CoroutineScope(coroutineContext).launch *{* (for my usecase), except that coroutineScope seems more elegant?

[Sam]

There’s a big difference in behaviour, because coroutineScope is a suspending function that waits for all the coroutines in its scope to complete. CoroutineScope(coroutineContext) creates a new scope but returns immediately. That means it breaks encapsulation and structured concurrency, because it lets you launch child coroutines that outlive the function (or flow) where they were created.
In general, you should always avoid calling CoroutineScope(coroutineContext) as it introduces a risk of leaking coroutines outside of their intended lifecycle.
In your original example, coroutineScope would actually not have worked as a replacement for CoroutineScope(coroutineContext) inside the onStart. It would delay the start of the flow indefinitely because it wouldn’t return until the side effect job was complete. But that’s because onStart is not the intended lifecycle of the side effect. The side effect is intended to run for the duration of the flow’s collection, which is why it makes sense to launch it inside the collect block.

[Radoslaw Juszczyk]

That means it breaks encapsulation and structured concurrency, because it lets you launch child coroutines that outlive the function (or flow) where they were created.

In what situation would they outlive it? Is that really possible considering that the newly created CoroutineScope uses parents corotuneContext?

[Sam]

You’re right that it can’t outlive the parent coroutineContext, since the new scope uses that context directly. What I meant was that it can outlive the current function or flow, which in itself can be a problem. An important rule of structured concurrency is that we can guarantee that when calling a flow or suspend function, the function or flow only returns/terminates once all of the coroutines it launched have also terminated.

[Radoslaw Juszczyk]

ahh got it

[Radoslaw Juszczyk]

and coroutineScope cannot outlive the current function since it suspends it

[Sam]

So an example of breaking structured concurrency would be something like

suspend fun bad() {
    CoroutineScope(currentCoroutineContext()).launch {
        delay(500)
        println("Still running!")
    }
}

It’s bad because it’s very confusing for anyone calling that function — the function appears to return immediately, but actually it’s still doing stuff

[Radoslaw Juszczyk]

got it! Thank you so much for taking time and giving these in depth explanations :)

[Radoslaw Juszczyk]

just one more thing:

coroutineScope {
        val sideEffect = launch {
            var c = 0
            //do stuff
            while (true) {
                delay(500)
                c++
                println("c = $c")
            }
        }
        onCompletion {
            sideEffect.cancel()
        }.collect {
            emit(it)
        }
    }

do we need the onCompletion { sideEffect.cancel() } then?
if collection ends then coroutineScope exits and its child jobs cancel, no?

[Sam]

Yes 👍 I added the onCompletion block to handle the case where the upstream flow is terminated. You said your upstream flow is a stateflow that never terminates, so in your case it’s not necessary.

[Sam]

I would probably still keep it though 😄 just in case. Code changes over time and somebody might later reuse the same code for a flow that does terminate…

[Radoslaw Juszczyk]

yes but if the collection ends (lets say it is not a StateFlow), then the coroutineScope ends too, doesnt it cancel its child jobs anyway?

[Sam]

Once the coroutineScope reaches the end of its lambda block it will actually just wait for all of its child jobs to complete. It doesn’t cancel them.

[Radoslaw Juszczyk]

aaah ok