Incorrect behavior of OnetimeWhileSubscribed with buffered Flow operators: upstream flow may be canceled before it has a chance to emit a single value

[cbeyls]

Library Version: 0.1.0

Describe the Bug:

The documentation of OnetimeWhileSubscribed says:

OnetimeWhileSubscribed is a SharingStarted strategy that ensures the upstream flow emits only once while a subscriber is active

This is incorrect. OnetimeWhileSubscribed only ensures that the coroutine collecting the upstream Flow is launched at most once, which is different.

Launching a coroutine once doesn't guarantee that the Flow it collects will have the opportunity to emit at least one value. If the coroutine is stopped before the Flow could emit at least one value (because of a slow HTTP request for example), then no value will ever be emitted because OnetimeWhileSubscribed will not start the coroutine a second time.

To go into technical details, even if the hasCollected flag is set after the suspending function emit(SharingCommand.START) returns, it doesn't mean that at that point a value has reached the StateFlow/SharedFlow cache downstream, because some Flow operators like buffer() or mapLatest() are buffered: in case of buffering, when emit() returns the emitted value is in a downstream buffer but has not yet been processed further down the chain.

In summary, the current implementation doesn't work reliably and should not be used.

Expected Behavior:

Instead of using a custom SharingStarted strategy like OnetimeWhileSubcribed acting upstream, the onetimeStateIn() implementation should collect the upstream Flow directly and only set the hasCollected flag after emitting the value directly to the downstream SharedFlow/StateFlow. This is the only reliable way to detect that the upstream Flow emitted at least one value and that this value has reached the SharedFlow/StateFlow cache.

Something like this:

fun <T> Flow<T>.onetimeStateIn(
    scope: CoroutineScope,
    started: SharingStarted,
    initialValue: T,
): StateFlow<T> {
    var hasCollected = false
    return flow {
        if (!hasCollected) {
            collect {
                emit(it)
                hasCollected = true
            }
        }
    }.stateIn(
        scope = scope,
        started = started,
        initialValue = initialValue
    )
}

Or better, set the flag after the upstream Flow completes (similarly to what the LiveData coroutine builder does):

fun <T> Flow<T>.onetimeStateIn(
    scope: CoroutineScope,
    started: SharingStarted,
    initialValue: T,
): StateFlow<T> {
    var hasCollected = false
    return flow {
        if (!hasCollected) {
            collect(this)
            hasCollected = true
        }
    }.stateIn(
        scope = scope,
        started = started,
        initialValue = initialValue,
    )
}

[skydoves]

Hey @cbeyls, thanks for sharing your thoughts. This API was initially designed as you mentioned: "OnetimeWhileSubscribed only ensures that the coroutine collecting the upstream Flow is launched at most once, which is different." I realize that my wording was unclear, and I appreciate you pointing that out. This is the intended behaviors.

My intention was to ensure that collecting values from the downstream could be easily managed by combining restartableStateIn with OnetimeWhileSubscribed. This approach allows the upstream coroutine to be relaunched in specific scenarios, such as handling bad network conditions, enabling both APIs to complement each other effectively.

That said, your perspective is also compelling, and I acknowledge that this API might introduce some confusion. Have you had a chance to try running this project with your solution and testing it with this unit test? I’d really appreciate any feedback or contributions if you have the opportunity. Thanks again for your valuable input!

[cbeyls]

I didn't try the above code with the unit tests but I don't think it would work as-is with the other operators, no. It was just an example to show that the detection for a successful flow completion must be done downstream after the end of the collection.

To be honest I think there are simpler ways to produce the same behavior as all the custom operators in this library.

The de facto pattern to restart a Flow is to use a trigger MutableStateFlow with mapLatest(), flatMapLatest() or transformLatest() and change its value:

private val triggerFlow = MutableStateFlow(0)

val resultFlow: Flow<UiState> = triggerFlow.flatMapLatest {
    myFlow()
}.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), UiState.Idle)

fun restart() {
    triggerFlow.update { it + 1 }
}

Then, if you also want the Flow to pause automatically after a successful completion so that it doesn't restart automatically when the screen becomes visible again, you can use a special value like -1 for the trigger. Here is an example which handles many cases:

private val triggerFlow = MutableStateFlow(0)

var isAutomaticPauseEnabled = true

val resultFlow: Flow<UiState> = triggerFlow.transformLatest { trigger ->
    if (trigger != -1) {
        emitAll(myFlow())
        if (isAutomaticPauseEnabled) {
            pause()
        }
    }
}.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), UiState.Idle)

fun pause() {
    triggerFlow.value = -1
}

fun resume() {
    triggerFlow.update { if (it != -1) it else 0 }
}

fun restart() {
    triggerFlow.update { it + 1 }
}

Now I suppose you could wrap the above logic into a custom operator but I don't think it's going to make the code much shorter.

[skydoves]

Hey @cbeyls, thanks for sharing this insightful perspective. I completely agree with you. Of course, there are many simpler ways to implement this from different angles, but the core responsibility of a library is to simplify such things—even though reducing one line of code, that has meaning at least.

That said, I’ll also explore a better solution for OnetimeWhileSubscribed, as you mentioned earlier. I believe there’s a more effective way to ensure that a coroutine launches from the upstream while emitting only once to the downstream. Let me take a deeper look into this—thanks again for the valuable input!