Properly close subscription on exception in Publisher.consumeEach

Author: qwwdfsad

reactive/kotlinx-coroutines-reactive/src/Channel.kt

@@ -31,9 +31,7 @@ public fun <T> Publisher<T>.openSubscription(request: Int = 0): ReceiveChannel<T
  * Subscribes to this [Publisher] and performs the specified action for each received element.
  */
 public suspend inline fun <T> Publisher<T>.consumeEach(action: (T) -> Unit) {
-    val channel = openSubscription()
-    for (x in channel) action(x)
-    channel.cancel()
+    openSubscription().consumeEach(action)
 }
 
 @Suppress("INVISIBLE_REFERENCE", "INVISIBLE_MEMBER")

reactive/kotlinx-coroutines-reactive/test/IntegrationTest.kt

@@ -74,6 +74,24 @@ class IntegrationTest(
         assertThat(cnt, IsEqual(1))
     }
 
+    @Test
+    fun testFailingConsumer() = runTest {
+        val pub = publish {
+            repeat(3) {
+                expect(it + 1) // expect(1), expect(2) *should* be invoked
+                send(it)
+            }
+        }
+
+        try {
+            pub.consumeEach {
+                throw TestException()
+            }
+        } catch (e: TestException) {
+            finish(3)
+        }
+    }
+
     @Test
     fun testNumbers() = runBlocking<Unit> {
         val n = 100 * stressTestMultiplier