Skip to content

fix: handle cancelled scope and empty flow in Flow.stateIn #4327

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Feb 12, 2025
Merged

fix: handle cancelled scope and empty flow in Flow.stateIn #4327

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
3b7e256
fix: rethrow CE when calling Flow.stateIn on cancelled scope
francescotescari Jan 6, 2025
80b114c
fix: throw NoSuchElementException on empty flow
francescotescari Jan 10, 2025
67a2166
fix: avoid cancelling scope on empty flow
francescotescari Feb 11, 2025
bcb1e14
Update kotlinx-coroutines-core/common/test/flow/sharing/StateInTest.kt
dkhalanskyjb Feb 12, 2025
ee11623
Undo the incorrect edit
dkhalanskyjb Feb 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion kotlinx-coroutines-core/common/src/flow/operators/Share.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -317,10 +317,11 @@ public fun <T> Flow<T>.stateIn(
* with multiple downstream subscribers. See the [StateFlow] documentation for the general concepts of state flows.
*
* @param scope the coroutine scope in which sharing is started.
* @throws NoSuchElementException if the upstream flow does not emit any value.
*/
public suspend fun <T> Flow<T>.stateIn(scope: CoroutineScope): StateFlow<T> {
val config = configureSharing(1)
val result = CompletableDeferred<StateFlow<T>>()
val result = CompletableDeferred<StateFlow<T>>(scope.coroutineContext[Job])
scope.launchSharingDeferred(config.context, config.upstream, result)
return result.await()
}
Expand All @@ -340,6 +341,9 @@ private fun <T> CoroutineScope.launchSharingDeferred(
}
}
}
if (state == null) {
result.completeExceptionally(NoSuchElementException("Flow is empty"))
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Notice: I think we shouldn't cancel the collecting scope here, as there is nothing wrong with the flow collection itself, the exception should be limited to stateIn IMHO.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, good point, let's not cancel the collecting scope. This is tricky to do ensure cleanly, though! Here's a slightly convoluted solution:

    /* If `scope`'s job is cancelled, both `deferredParentJob` and the deferred will be cancelled.
    If the deferred is cancelled, this will not cancel `deferredParentJob` (as it's a `SupervisorJob`),
    and `CoroutineExceptionHandler` present in `scope` (if any) won't get invoked,
    as `Deferred` values are themselves responsible for propagating their exceptions. */
    val deferredParentJob = SupervisorJob(scope.coroutineContext[Job])
    val result = CompletableDeferred<StateFlow<T>>(deferredParentJob)
    scope.launchSharingDeferred(config.context, config.upstream, result)
    return try {
        result.await()
    } finally {
        deferredParentJob.cancel()
    }

then the modified test is:

    @Test
    fun testThrowsNoSuchElementExceptionOnEmptyFlow() = runTest {
        val flow = emptyFlow<Any>()
        assertFailsWith<NoSuchElementException> {
            flow.stateIn(this)
        }
        assertEquals(true, coroutineContext[Job]?.isActive)
        coroutineContext.cancelChildren()
    }

(Note this runTest is not kotlinx.coroutines.test.runTest!) If the main coroutine of runTest gets cancelled, the whole test fails; also, if there are any uncaught exceptions in child coroutines of runTest, they do get reported at the end of the test. So, this test does check that NoSuchElementException doesn't propagate upwards.

The only other option I see is to rely on our internal low-level job.invokeOnCompletion function to avoid the parent-child relationship between scope and result altogether. After implementing that option, I don't like how it recreates parts of child cancellation logic. This looks like a weird micro-optimization.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Another simpler (not elegant) trick is to make CompletableDeferred of type CompletableDeferred<Result<StateFlow<T>>> and complete the deferred successfully but with Result.failure and unpack with getOrThrow it stateIn side.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good idea!

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@francescotescari, would you like to implement this, or should I do it?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry for the delay, addressed it now.
LMK if it looks good to you 🙏

}
} catch (e: Throwable) {
// Notify the waiter that the flow has failed
result.completeExceptionally(e)
Expand Down
18 changes: 18 additions & 0 deletions kotlinx-coroutines-core/common/test/flow/sharing/StateInTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package kotlinx.coroutines.flow
import kotlinx.coroutines.testing.*
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlin.coroutines.*
import kotlin.test.*

/**
Expand Down Expand Up @@ -88,4 +89,21 @@ class StateInTest : TestBase() {
fun testSubscriptionByFirstSuspensionInStateFlow() = runTest {
testSubscriptionByFirstSuspensionInCollect(flowOf(1).stateIn(this@runTest)) { }
}

@Test
fun testRethrowsCEOnCancelledScope() = runTest {
val cancelledScope = CoroutineScope(EmptyCoroutineContext).apply { cancel("CancelMessageToken") }
val flow = flowOf(1, 2, 3)
assertFailsWith<CancellationException>("CancelMessageToken") {
flow.stateIn(cancelledScope)
}
}

@Test
fun testThrowsNoSuchElementExceptionOnEmptyFlow() = runTest {
val flow = emptyFlow<Any>()
assertFailsWith<NoSuchElementException> {
flow.stateIn(GlobalScope)
}
}
}