Fix Flow.timeout swallowing the channel closure exception (#4072)

Fixes #4071
Kotlin · Mar 25, 2024 · 0ca7358 · 0ca7358
1 parent 60d2fe8
commit 0ca7358
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 0 deletions.
diff --git a/kotlinx-coroutines-core/common/src/flow/operators/Delay.kt b/kotlinx-coroutines-core/common/src/flow/operators/Delay.kt
@@ -394,6 +394,7 @@ private fun <T> Flow<T>.timeoutInternal(
             value.onSuccess {
                 downStream.emit(it)
             }.onClosed {
+                it?.let { throw it }
                 return@onReceiveCatching false
             }
             return@onReceiveCatching true

diff --git a/kotlinx-coroutines-core/common/test/flow/operators/TimeoutTest.kt b/kotlinx-coroutines-core/common/test/flow/operators/TimeoutTest.kt
@@ -237,6 +237,17 @@ class TimeoutTest : TestBase() {
         testImmediateTimeout(-1.seconds)
     }
 
+    @Test
+    fun testClosing() = runTest {
+        assertFailsWith<TestException> {
+            channelFlow<Int> { close(TestException()) }
+                .timeout(Duration.INFINITE)
+                .collect {
+                    expectUnreached()
+                }
+        }
+    }
+
     private fun testImmediateTimeout(timeout: Duration) {
         expect(1)
         val flow = emptyFlow<Int>().timeout(timeout)