Fix exception types for channels to ensure transparency & reporting

* ReceiveChannel.cancel always closes channel with CancellationException,
  so sending or receiving from a cancelled channel produces the
  corresponding CancellationException.
* Cancelling produce builder has similar effect, but an more specific
  instance of JobCancellationException is created.
* This ensure that produce/consumeEach pair is transparent with respect
  to cancellation and can be used to build "identity" transformation
  of the flow (the corresponding test is added).
* ClosedSendChannelException is now a subclass of IllegalStateException,
  so that trying to send to a channel that was closed normally is
  reported as program error and is not eaten (test is added).

Fixes #957
Fixes #1128

Author: elizarov

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

@@ -669,7 +669,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
         cancelInternal(cause)
 
     final override fun cancel(cause: CancellationException?) {
-        cancelInternal(cause)
+        cancelInternal(cause ?: CancellationException("$classSimpleName was cancelled"))
     }
 
     // It needs to be internal to support deprecated cancel(Throwable?) API

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

@@ -242,14 +242,16 @@ public interface ReceiveChannel<out E> {
     /**
      * Cancels reception of remaining elements from this channel with an optional [cause].
      * This function closes the channel and removes all buffered sent elements from it.
+     *
      * A cause can be used to specify an error message or to provide other details on
      * a cancellation reason for debugging purposes.
+     * If cause is not specified, then an instance of [CancellationException] with a
+     * default message is create to [close][SendChannel.close] the channel. 
      *
      * Immediately after invocation of this function [isClosedForReceive] and
      * [isClosedForSend][SendChannel.isClosedForSend]
-     * on the side of [SendChannel] start returning `true`, so all attempts to send to this channel
-     * afterwards will throw [ClosedSendChannelException], while attempts to receive will throw
-     * [ClosedReceiveChannelException].
+     * on the side of [SendChannel] start returning `true`. All attempts to send to this channel
+     * or receive from this channel will throw [CancellationException].
      */
     public fun cancel(cause: CancellationException? = null)
 
@@ -382,14 +384,18 @@ public fun <E> Channel(capacity: Int = RENDEZVOUS): Channel<E> =
  * Indicates attempt to [send][SendChannel.send] on [isClosedForSend][SendChannel.isClosedForSend] channel
  * that was closed without a cause. A _failed_ channel rethrows the original [close][SendChannel.close] cause
  * exception on send attempts.
+ *
+ * This exception is a subclass of [IllegalStateException] because, conceptually, sender is responsible
+ * for closing the channel and not be trying to send anything after the channel was close. Attempts to
+ * send into the closed channel indicate logical error in the sender's code.
  */
-public class ClosedSendChannelException(message: String?) : CancellationException(message)
+public class ClosedSendChannelException(message: String?) : IllegalStateException(message)
 
 /**
  * Indicates attempt to [receive][ReceiveChannel.receive] on [isClosedForReceive][ReceiveChannel.isClosedForReceive]
  * channel that was closed without a cause. A _failed_ channel rethrows the original [close][SendChannel.close] cause
  * exception on receive attempts.
  *
- * This exception is subclass of [NoSuchElementException] to be consistent with plain collections.
+ * This exception is a subclass of [NoSuchElementException] to be consistent with plain collections.
  */
-public class ClosedReceiveChannelException(message: String?) : NoSuchElementException(message)
+public class ClosedReceiveChannelException(message: String?) : NoSuchElementException(message)

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

@@ -28,8 +28,15 @@ internal open class ChannelCoroutine<E>(
     }
 
     override fun cancelInternal(cause: Throwable?): Boolean {
-        _channel.cancel(cause?.toCancellationException()) // cancel the channel
-        cancelCoroutine(cause) // cancel the job
+        val exception = cause?.toCancellationException()
+            ?: JobCancellationException("Job was cancelled", null, this)
+        _channel.cancel(exception) // cancel the channel
+        cancelCoroutine(exception) // cancel the job
         return true // does not matter - result is used in DEPRECATED functions only
     }
+
+    @Suppress("UNCHECKED_CAST")
+    suspend fun sendFair(element: E) {
+        (_channel as AbstractSendChannel<E>).sendFair(element)
+    }
 }

kotlinx-coroutines-core/common/src/flow/internal/AbortFlowException.kt

@@ -0,0 +1,16 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.flow.internal
+
+import kotlinx.coroutines.*
+
+/**
+ * This exception is thrown when operator need no more elements from the flow.
+ * This exception should never escape outside of operator's implementation.
+ */
+internal class AbortFlowException : CancellationException("Flow was aborted, no more elements needed") {
+    // TODO expect/actual
+    // override fun fillInStackTrace(): Throwable = this
+}

kotlinx-coroutines-core/common/src/flow/operators/Limit.kt

@@ -8,9 +8,10 @@
 
 package kotlinx.coroutines.flow
 
-import kotlinx.coroutines.flow.unsafeFlow as flow
 import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.internal.*
 import kotlin.jvm.*
+import kotlinx.coroutines.flow.unsafeFlow as flow
 
 /**
  * Returns a flow that ignores first [count] elements.
@@ -57,10 +58,10 @@ public fun <T> Flow<T>.take(count: Int): Flow<T> {
             collect { value ->
                 emit(value)
                 if (++consumed == count) {
-                    throw TakeLimitException()
+                    throw AbortFlowException()
                 }
             }
-        } catch (e: TakeLimitException) {
+        } catch (e: AbortFlowException) {
             // Nothing, bail out
         }
     }
@@ -74,14 +75,9 @@ public fun <T> Flow<T>.takeWhile(predicate: suspend (T) -> Boolean): Flow<T> = f
     try {
         collect { value ->
             if (predicate(value)) emit(value)
-            else throw TakeLimitException()
+            else throw AbortFlowException()
         }
-    } catch (e: TakeLimitException) {
+    } catch (e: AbortFlowException) {
         // Nothing, bail out
     }
 }
-
-private class TakeLimitException : CancellationException("Flow limit is reached, cancelling") {
-    // TODO expect/actual
-    // override fun fillInStackTrace(): Throwable = this
-}

kotlinx-coroutines-core/common/src/flow/operators/Zip.kt

@@ -78,7 +78,7 @@ public fun <T1, T2, R> Flow<T1>.combineLatest(other: Flow<T2>, transform: suspen
 
 private inline fun SelectBuilder<Unit>.onReceive(
     isClosed: Boolean,
-    channel: Channel<Any>,
+    channel: ReceiveChannel<Any>,
     crossinline onClosed: () -> Unit,
     noinline onReceive: suspend (value: Any) -> Unit
 ) {
@@ -90,18 +90,11 @@ private inline fun SelectBuilder<Unit>.onReceive(
 }
 
 // Channel has any type due to onReceiveOrNull. This will be fixed after receiveOrClosed
-private fun CoroutineScope.asFairChannel(flow: Flow<*>): Channel<Any> {
-    val channel = RendezvousChannel<Any>() // Explicit type
-    launch {
-        try {
-            flow.collect { value ->
-                channel.sendFair(value ?: NullSurrogate)
-            }
-        } finally {
-            channel.close()
-        }
+private fun CoroutineScope.asFairChannel(flow: Flow<*>): ReceiveChannel<Any> = produce {
+    val channel = channel as ChannelCoroutine<Any>
+    flow.collect { value ->
+        channel.sendFair(value ?: NullSurrogate)
     }
-    return channel
 }
 
 
@@ -133,7 +126,9 @@ public fun <T1, T2, R> Flow<T1>.zip(other: Flow<T2>, transform: suspend (T1, T2)
          *
          * Invariant: this clause is invoked only when all elements from the channel were processed (=> rendezvous restriction).
          */
-        (second as SendChannel<*>).invokeOnClose { first.cancel() }
+        (second as SendChannel<*>).invokeOnClose {
+            first.cancel(AbortFlowException())
+        }
 
         val otherIterator = second.iterator()
         try {
@@ -144,8 +139,10 @@ public fun <T1, T2, R> Flow<T1>.zip(other: Flow<T2>, transform: suspend (T1, T2)
                 val secondValue = NullSurrogate.unbox<T2>(otherIterator.next())
                 emit(transform(NullSurrogate.unbox(value), NullSurrogate.unbox(secondValue)))
             }
+        } catch (e: AbortFlowException) {
+            // complete
         } finally {
-            second.cancel()
+            second.cancel(AbortFlowException())
         }
     }
 }

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

@@ -170,23 +170,25 @@ class ArrayBroadcastChannelTest : TestBase() {
         // start consuming
         val sub = channel.openSubscription()
         var expected = 0
-        sub.consumeEach {
-            check(it == ++expected)
-            if (it == 2) {
-                sub.cancel()
+        assertFailsWith<CancellationException> {
+            sub.consumeEach {
+                check(it == ++expected)
+                if (it == 2) {
+                    sub.cancel()
+                }
             }
         }
         check(expected == 2)
     }
 
     @Test
-    fun testReceiveFromClosedSub() = runTest({ it is ClosedReceiveChannelException }) {
+    fun testReceiveFromCancelledSub() = runTest {
         val channel = BroadcastChannel<Int>(1)
         val sub = channel.openSubscription()
         assertFalse(sub.isClosedForReceive)
         sub.cancel()
         assertTrue(sub.isClosedForReceive)
-        sub.receive()
+        assertFailsWith<CancellationException> { sub.receive() }
     }
 
     @Test

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

@@ -134,7 +134,7 @@ class ArrayChannelTest : TestBase() {
         q.cancel()
         check(q.isClosedForSend)
         check(q.isClosedForReceive)
-        check(q.receiveOrNull() == null)
+        assertFailsWith<CancellationException> { q.receiveOrNull() }
         finish(12)
     }
 

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

@@ -8,7 +8,6 @@ import kotlinx.coroutines.*
 import kotlin.test.*
 
 class BasicOperationsTest : TestBase() {
-
     @Test
     fun testSimpleSendReceive() = runTest {
         // Parametrized common test :(
@@ -20,6 +19,11 @@ class BasicOperationsTest : TestBase() {
         TestChannelKind.values().forEach { kind -> testOffer(kind) }
     }
 
+    @Test
+    fun testSendAfterClose() = runTest {
+        TestChannelKind.values().forEach { kind -> testSendAfterClose(kind) }
+    }
+
     @Test
     fun testReceiveOrNullAfterClose() = runTest {
         TestChannelKind.values().forEach { kind -> testReceiveOrNull(kind) }
@@ -128,6 +132,23 @@ class BasicOperationsTest : TestBase() {
         d.await()
     }
 
+    /**
+     * [ClosedSendChannelException] should not be eaten.
+     * See [https://github.com/Kotlin/kotlinx.coroutines/issues/957]
+     */
+    private suspend fun testSendAfterClose(kind: TestChannelKind) {
+        assertFailsWith<ClosedSendChannelException> {
+            coroutineScope {
+                val channel = kind.create()
+                channel.close()
+
+                launch {
+                    channel.send(1)
+                }.join()
+            }
+        }
+    }
+
     private suspend fun testSendReceive(kind: TestChannelKind, iterations: Int) = coroutineScope {
         val channel = kind.create()
         launch {

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

@@ -73,7 +73,7 @@ class ConflatedChannelTest : TestBase() {
         q.cancel()
         check(q.isClosedForSend)
         check(q.isClosedForReceive)
-        check(q.receiveOrNull() == null)
+        assertFailsWith<CancellationException> { q.receiveOrNull() }
         finish(2)
     }
 

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

@@ -31,7 +31,7 @@ class LinkedListChannelTest : TestBase() {
         q.cancel()
         check(q.isClosedForSend)
         check(q.isClosedForReceive)
-        check(q.receiveOrNull() == null)
+        assertFailsWith<CancellationException> { q.receiveOrNull() }
     }
 
     @Test

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

@@ -38,7 +38,7 @@ class ProduceTest : TestBase() {
                 expectUnreached()
             } catch (e: Throwable) {
                 expect(7)
-                check(e is ClosedSendChannelException)
+                check(e is CancellationException)
                 throw e
             }
             expectUnreached()
@@ -48,7 +48,7 @@ class ProduceTest : TestBase() {
         expect(4)
         c.cancel()
         expect(5)
-        assertNull(c.receiveOrNull())
+        assertFailsWith<CancellationException> { c.receiveOrNull() }
         expect(6)
         yield() // to produce
         finish(8)

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

@@ -272,7 +272,7 @@ class RendezvousChannelTest : TestBase() {
         q.cancel()
         check(q.isClosedForSend)
         check(q.isClosedForReceive)
-        check(q.receiveOrNull() == null)
+        assertFailsWith<CancellationException> { q.receiveOrNull() }
         finish(12)
     }
 

kotlinx-coroutines-core/common/test/flow/IdFlowTest.kt

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+@file:Suppress("NAMED_ARGUMENTS_NOT_ALLOWED") // KT-21913
+
+package kotlinx.coroutines.flow
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.*
+import kotlin.test.*
+
+// See https://github.com/Kotlin/kotlinx.coroutines/issues/1128
+class IdFlowTest : TestBase() {
+    @Test
+    fun testCancelInCollect() = runTest(
+        expected = { it is CancellationException }
+    ) {
+        expect(1)
+        flow {
+            expect(2)
+            emit(1)
+            expect(3)
+            delay(Long.MAX_VALUE)
+            expectUnreached()
+        }.idScoped().collect { value ->
+            expect(4)
+            assertEquals(1, value)
+            kotlin.coroutines.coroutineContext.cancel()
+            finish(5)
+        }
+        expectUnreached()
+    }
+
+    @Test
+    fun testCancelInFlow() = runTest(
+        expected = { it is CancellationException }
+    ) {
+        expect(1)
+        flow {
+            expect(2)
+            emit(1)
+            kotlin.coroutines.coroutineContext.cancel()
+            expect(3)
+        }.idScoped().collect { value ->
+            finish(4)
+            assertEquals(1, value)
+        }
+        expectUnreached()
+    }
+}
+
+/**
+ * This flow should be "identity" function with respect to cancellation.
+ */
+private fun <T> Flow<T>.idScoped(): Flow<T> = flow {
+    coroutineScope {
+        val channel = produce<T> {
+            collect { send(it) }
+        }
+        channel.consumeEach {
+            emit(it)
+        }
+    }
+}

kotlinx-coroutines-core/jvm/test/channels/TickerChannelCommonTest.kt

@@ -48,8 +48,7 @@ class TickerChannelCommonTest(private val channelFactory: Channel) : TestBase()
 
             delayChannel.cancel()
             delay(5100)
-            delayChannel.checkEmpty()
-            delayChannel.cancel()
+            assertFailsWith<CancellationException> { delayChannel.poll() }
         }
     }