Specify the behavior of Channel.consumeEach on scope cancellation

Author: globsterg

kotlinx-coroutines-core/common/src/channels/Channels.common.kt

@@ -107,7 +107,8 @@ public inline fun <E, R> ReceiveChannel<E>.consume(block: ReceiveChannel<E>.() -
  * Performs the given [action] for each received element and [cancels][ReceiveChannel.cancel] the channel afterward.
  *
  * This function stops processing elements when the channel is closed,
- * the coroutine in which the collection is performed gets cancelled,
+ * the coroutine in which the collection is performed gets cancelled and there are no readily available elements in the
+ * channel's buffer,
  * or an early return from [action] happens.
  * Throwing an exception from [action] will attempt to close the channel using the thrown exception.
  *

kotlinx-coroutines-core/common/test/channels/ConsumeTest.kt

@@ -91,6 +91,36 @@ class ConsumeTest: TestBase() {
         assertTrue(channel.isClosedForReceive)
     }
 
+    /** Checks that [ReceiveChannel.consumeEach] reacts to cancellation, but processes the elements that are
+     * readily available in the buffer. */
+    @Test
+    fun testConsumeEachExitsOnCancellation() = runTest {
+        val undeliveredElements = mutableListOf<Int>()
+        val channel = Channel<Int>(2, onUndeliveredElement = {
+            undeliveredElements.add(it)
+        })
+        launch {
+            // These two elements will be sent and put into the buffer:
+            channel.send(0)
+            channel.send(1)
+            // This element will not fit into the buffer, so `send` suspends:
+            channel.send(2)
+            // At this point, the consumer's `launch` is cancelled.
+            yield() // Allow the cancellation handler of the consumer to run.
+            // Try to send a new element, which will fail at this point:
+            channel.send(3)
+            fail("unreached")
+        }
+        launch {
+            channel.consumeEach {
+                cancel()
+                assertTrue(it in 0..2)
+            }
+        }.join()
+        assertTrue(channel.isClosedForReceive)
+        assertEquals(listOf(3), undeliveredElements)
+    }
+
     /** Check that [BroadcastChannel.consume] does not suffer from KT-58685 */
     @Suppress("DEPRECATION", "DEPRECATION_ERROR")
     @Test