I have a view emitting a flow of Actions which is collected by a handler which emits a flow of Mutations (that will update the state) and they have different lifecycle scopes (the handler can outlive the views). How can I do it so that I can cancel the view’s scope (no more Actions come through) yet the handling of the actions still being processed is not cancelled? Say the handler takes 5 seconds processing each action. When I cancel the view scope, I know that I should expect more mutations to come through but they get cancelled too. I’ve gone through the docs and tried many different alternatives but none worked.

We use a

StateFlow

or an

EventFlow

for that. We rolled our own

StateFlow

and

EventFlow

, and they are `Flow`s that are backed by a

ConflatedBroadcastChannel

(for StateFlow) or a plain rendezvous

Channel

(for EventFlow).

I use a state flow in other parts of the system, but I don’t think that would do it in this case. I can’t use conflated flows here. Each value is important (each button click; each update (mutation) on the data). I don’t want just the latest one. The handler itself knows how to react to them but is stateless. It emits mutations which are fed to a reducer and then it updates the state flow… But the handler doesn’t do that directly. The handler only receives actions, does what it needs to do, and emits mutations.

Then an EventFlow would work, correct? It would not cache values. E.g:

class EventFlow<T> private constructor(
    private val channel: Channel<T>
): Flow<T> by channel.receiveAsFlow() {

    suspend fun sendValue(value: T) {
        channel.send(value)
    }

    companion object {
        operator fun <T> invoke(): EventFlow<T> = EventFlow(Channel())
    }
}

(note;

receiveAsFlow

produces a Flow that can only have one collector/observer. you’d need to to a bit more code to allow for more than one collector)

one collector is fine 👍 is there some article explaining that / how it works? I googled EventFlow and it took me to posts about shared flow which hasn’t been implemented (officially at least) AFAIK.

Yup, StateFlow is in kotlin 1.3.7, but no EventFlow yet. We rolled our own on both of them (for now; since they need to work with Kotlin Multi Platform Multi Threading).

hmmm the rendezvous could become an issue but this is a step forward… but it still doesn’t work for me. Is this code right?

val eventFlow = EventFlow<Action>()
            viewScope.launch {
                viewActions
                    .buffer()
                    .onEach {
//                        log("sending action $it")
                        eventFlow.sendValue(it)
                    }
                    .launchIn(viewScope)
            }
            handlerScope.launch {
                eventFlow
                    .collect { action ->
                        log("Handling action $action")
                        handlerScope.launch { 
                            handler.handle(action)
                                .collect {
                                    // do something with mutation
                                    log("Mutation $it")
                                    counter++
                                }
                        }
                    }
            }

The

viewScope

is the scope of the producer. The

handleScope

is the scope of the consumer. Cancelling the viewScope will just stop sending values. It won’t cancel the handleScope, but the call to

eventFlow.collect

will not do anything, since there is no-one sending any values….

the handler is both a consumer of actions and a producer of mutations

What happens if you remove the second

handlerScope.launch

call (and just leave the

log("Handling action $action")

statement? I expect that

log

statement to execute each time someone is calling

eventFlow.sendValue(…)

during the entire lifetime of

handlerScope

. If you cancel the

viewScope

, the

eventFlow.sendValue

won’t be called anymore. This means that the

log

statement won’t get called anymore either, even though the

handlerScope

is still active.

I wonder if this is the problem in my test

val viewScope = CoroutineScope(newCoroutineContext(coroutineContext))
            val handlerScope = CoroutineScope(newCoroutineContext(coroutineContext))

I assume that by cancelling one scope, the other shouldn’t be cancelled. The funny thing is my original code appears to work in the real application but I can’t write unit tests to verify the behaviour

The seem to share a coroutineContext… if that one has a

Job

, they share that

Job

And cancelling one then cancels the other

🤦

Just do a

= CoroutineScope(myDispatcher)

for each one of them.

CoroutineScope(TestCoroutineDispatcher())

seems to solve my problem of killing both scopes. Thanks for that, that was my actual problem! The rest of the code is still not working but now I have more to explore! Thanks