Hi all, Let's say I have a state flow created usin...

# coroutines

[Naveen Perumal]

Hi all, Let's say I have a state flow created using stateIn(scope). When scope is cancelled, the flow will be completed and no more emissions can happen. In that case why are the collectors not cancelled? Consider the following snippet,

coroutineScope {
        val flow = flow {
            emit(1)
            delay(1000)
            emit(2)
            delay(1000)
            emit(3)
            delay(1000)
            emit(4)
        }

        val stateFlowScope = CoroutineScope(Job() + Dispatchers.Default)
        val stateFlow = flow
            .onCompletion { println("State flow completed") }
            .stateIn(stateFlowScope)

        val flowCollectorJob = launch {
            stateFlow
                .onCompletion { "Collector completed" }
                .collect { value ->
                    println("StateFlow value: $value")
                }
        }

        delay(2000)
        println("Cancelling scope...")
        stateFlowScope.cancel()
        flowCollectorJob.join()
        println("-------COMPLETED-------")
    }

It produces the following output

StateFlow value: 1
StateFlow value: 2
Cancelling scope...
State flow completed

The collector is not completed. Why is that and is there any way to stop the collector when the flow completes?

[ross_a]

There isn't unfortunately
StateFlow is a specialisation of SharedFlow and per the docs "Shared flow never completes", it's tricky because I too wish it did.

[ross_a]

Thought a bit more and this might help:

val flow = flowOf<String>()

sealed interface MaterializedEvent<out T> {
    @JvmInline
    value class Update<T>(val t: T) : MaterializedEvent<T>

    @JvmInline
    value class Completed(val throwable: Throwable?) : MaterializedEvent<Nothing>
}

fun <T> Flow<T>.stateInMaterialized(scope: CoroutineScope): StateFlow<MaterializedEvent<T>> {
    val state = MutableStateFlow<MaterializedEvent<T>?>(null)

    this.onEach { state.value = Update(it) }
        .onCompletion { state.value = Completed(it) }
        .launchIn(scope)

    // or only complete if the scope does
//    scope.coroutineContext[Job]?.invokeOnCompletion { state.value = Completed(it) }

    return state.drop(1) as StateFlow<MaterializedEvent<T>>
}

fun <T> Flow<MaterializedEvent<T>>.dematerialize(): Flow<T> =
    transformWhile { valueOrCompletion ->
        when (valueOrCompletion) {
            is Update -> {
                emit(valueOrCompletion.t)
                true
            }
            is Completed -> {
                valueOrCompletion.throwable?.let { throw it }
                false
            }
        }
    }

[Naveen Perumal]

Thanks for the reply.
This solution works as expected. Just a small correction. state.drop(1).stateIn(scope) should be used instead of state.drop(1) .
And I'm curious to know if there is any specific reason as to why the collectors aren't cancelled automatically when the scope of the flow itself is cancelled.

[ross_a]

No problem, I haven't actually tested it yet to find out the unexpected so I don't know I'm afraid

[ross_a]

ah yes it's not a stateflow after the drop 😕

[Naveen Perumal]

ok. thanks

[ross_a]

fun <T> Flow<T>.stateInMaterialized(scope: CoroutineScope, initialValue: T): StateFlow<MaterializedEvent<T>> {
    val state = MutableStateFlow<MaterializedEvent<T>>(Update(initialValue))

    this.onEach { state.value = Update(it) }
        .onCompletion { state.value = Completed(it) }
        .launchIn(scope)

    // or only complete if the scope does
//    scope.coroutineContext[Job]?.invokeOnCompletion { state.value = Completed(it) }

    return state
}

Probably makes more sense to do this then, and create a suspending version if you don't want the initial value

[Naveen Perumal]

state.drop(1).stateIn(scope) should be used or else the collector won't complete since state isn't scoped to the provided scope.

[ross_a]

which collector sorry? the one emitting to the MutableStateFlow is via the launchIn

[ross_a]

the MutableStateFlow doesn't need to be scoped, its collectors are scoped to whatever they are launched in

[Naveen Perumal]

val stateFlow = flow
            .onCompletion { println("State flow completed") }
            .stateInMaterialized(stateFlowScope)

The collectors of stateFlow seem to be collecting forever whenever scope is cancelled if state is returned directly. Whereas the collectors seem to be completed on cancellation of scope if state.drop(1).stateIn(scope) is returned.

[ross_a]

you need to dematerialize it for the cancellation to be unwrapped and actually occur

[ross_a]

val stateFlow = flow
            .onCompletion { println("State flow completed") }
            .stateInMaterialized(stateFlowScope)

val collector = stateFlow.dematerialize().collect {}

[Naveen Perumal]

sorry. my bad. Didn't check correctly. It works as expected