Document that SharedFlow.collect subscribes by its first suspension (#3885)

Author: dkhalanskyjb

kotlinx-coroutines-core/common/src/flow/SharedFlow.kt

@@ -137,6 +137,16 @@ public interface SharedFlow<out T> : Flow<T> {
      * **A shared flow never completes**. A call to [Flow.collect] or any other terminal operator
      * on a shared flow never completes normally.
      *
+     * It is guaranteed that, by the time the first suspension happens, [collect] has already subscribed to the
+     * [SharedFlow] and is eligible for receiving emissions. In particular, the following code will always print `1`:
+     * ```
+     * val flow = MutableSharedFlow<Int>()
+     * launch(start = CoroutineStart.UNDISPATCHED) {
+     *   flow.collect { println(1) }
+     * }
+     * flow.emit(1)
+     * ```
+     *
      * @see [Flow.collect] for implementation and inheritance details.
      */
     override suspend fun collect(collector: FlowCollector<T>): Nothing

kotlinx-coroutines-core/common/test/flow/sharing/ShareInTest.kt

@@ -236,4 +236,9 @@ class ShareInTest : TestBase() {
         assertEquals(239, shared.first())
         j.cancel()
     }
+
+    @Test
+    fun testSubscriptionByFirstSuspensionInSharedFlow() = runTest {
+        testSubscriptionByFirstSuspensionInCollect(flowOf(1).stateIn(this@runTest), emit = { })
+    }
 }

kotlinx-coroutines-core/common/test/flow/sharing/SharedFlowTest.kt

@@ -818,4 +818,24 @@ class SharedFlowTest : TestBase() {
         j2.cancelAndJoin()
         assertEquals(0, flow.subscriptionCount.first())
     }
+
+    @Test
+    fun testSubscriptionByFirstSuspensionInSharedFlow() = runTest {
+        testSubscriptionByFirstSuspensionInCollect(MutableSharedFlow()) { emit(it) }
+    }
+}
+
+/**
+ * Check that, by the time [SharedFlow.collect] suspends for the first time, its subscription is already active.
+ */
+inline fun<T: Flow<Int>> CoroutineScope.testSubscriptionByFirstSuspensionInCollect(flow: T, emit: T.(Int) -> Unit) {
+    var received = 0
+    val job = launch(start = CoroutineStart.UNDISPATCHED) {
+        flow.collect {
+            received = it
+        }
+    }
+    flow.emit(1)
+    assertEquals(1, received)
+    job.cancel()
 }

kotlinx-coroutines-core/common/test/flow/sharing/StateFlowTest.kt

@@ -181,4 +181,9 @@ class StateFlowTest : TestBase() {
         state.update { it + 3 }
         assertEquals(5, state.value)
     }
+
+    @Test
+    fun testSubscriptionByFirstSuspensionInStateFlow() = runTest {
+        testSubscriptionByFirstSuspensionInCollect(MutableStateFlow(0)) { value = it; yield() }
+    }
 }

kotlinx-coroutines-core/common/test/flow/sharing/StateInTest.kt

@@ -86,4 +86,9 @@ class StateInTest : TestBase() {
         assertFailsWith<TestException> { flow.stateIn(CoroutineScope(currentCoroutineContext() + Job() + ceh)) }
         finish(3)
     }
+
+    @Test
+    fun testSubscriptionByFirstSuspensionInStateFlow() = runTest {
+        testSubscriptionByFirstSuspensionInCollect(flowOf(1).stateIn(this@runTest)) { }
+    }
 }