Add job property to ReceiveChannel and SendChannel

Fixes #260

Author: qwwdfsad

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt

@@ -456,11 +456,12 @@ public final class kotlinx/coroutines/experimental/YieldKt {
 }
 
 public abstract class kotlinx/coroutines/experimental/channels/AbstractChannel : kotlinx/coroutines/experimental/channels/AbstractSendChannel, kotlinx/coroutines/experimental/channels/Channel {
-	public fun <init> ()V
+	public fun <init> (Lkotlinx/coroutines/experimental/Job;)V
 	public fun cancel (Ljava/lang/Throwable;)Z
 	protected fun cleanupSendQueueOnCancel ()V
 	protected final fun describeTryPoll ()Lkotlinx/coroutines/experimental/channels/AbstractChannel$TryPollDesc;
 	protected final fun getHasReceiveOrClosed ()Z
+	public final fun getJob ()Lkotlinx/coroutines/experimental/Job;
 	public final fun getOnReceive ()Lkotlinx/coroutines/experimental/selects/SelectClause1;
 	public final fun getOnReceiveOrNull ()Lkotlinx/coroutines/experimental/selects/SelectClause1;
 	protected abstract fun isBufferAlwaysEmpty ()Z
@@ -539,15 +540,18 @@ public abstract interface class kotlinx/coroutines/experimental/channels/ActorSc
 }
 
 public final class kotlinx/coroutines/experimental/channels/ArrayBroadcastChannel : kotlinx/coroutines/experimental/channels/AbstractSendChannel, kotlinx/coroutines/experimental/channels/BroadcastChannel {
-	public fun <init> (I)V
+	public fun <init> (ILkotlinx/coroutines/experimental/Job;)V
+	public synthetic fun <init> (ILkotlinx/coroutines/experimental/Job;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public fun close (Ljava/lang/Throwable;)Z
 	public final fun getCapacity ()I
+	public fun getJob ()Lkotlinx/coroutines/experimental/Job;
 	public fun open ()Lkotlinx/coroutines/experimental/channels/SubscriptionReceiveChannel;
 	public fun openSubscription ()Lkotlinx/coroutines/experimental/channels/SubscriptionReceiveChannel;
 }
 
 public class kotlinx/coroutines/experimental/channels/ArrayChannel : kotlinx/coroutines/experimental/channels/AbstractChannel {
-	public fun <init> (I)V
+	public fun <init> (ILkotlinx/coroutines/experimental/Job;)V
+	public synthetic fun <init> (ILkotlinx/coroutines/experimental/Job;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	protected fun cleanupSendQueueOnCancel ()V
 	protected fun getBufferDebugString ()Ljava/lang/String;
 	public final fun getCapacity ()I
@@ -712,7 +716,10 @@ public final class kotlinx/coroutines/experimental/channels/ConflatedBroadcastCh
 	public static final field UNDEFINED Lkotlinx/coroutines/experimental/internal/Symbol;
 	public fun <init> ()V
 	public fun <init> (Ljava/lang/Object;)V
+	public fun <init> (Lkotlinx/coroutines/experimental/Job;)V
+	public synthetic fun <init> (Lkotlinx/coroutines/experimental/Job;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public fun close (Ljava/lang/Throwable;)Z
+	public fun getJob ()Lkotlinx/coroutines/experimental/Job;
 	public fun getOnSend ()Lkotlinx/coroutines/experimental/selects/SelectClause2;
 	public final fun getValue ()Ljava/lang/Object;
 	public final fun getValueOrNull ()Ljava/lang/Object;
@@ -726,6 +733,8 @@ public final class kotlinx/coroutines/experimental/channels/ConflatedBroadcastCh
 
 public class kotlinx/coroutines/experimental/channels/ConflatedChannel : kotlinx/coroutines/experimental/channels/AbstractChannel {
 	public fun <init> ()V
+	public fun <init> (Lkotlinx/coroutines/experimental/Job;)V
+	public synthetic fun <init> (Lkotlinx/coroutines/experimental/Job;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	protected final fun isBufferAlwaysEmpty ()Z
 	protected final fun isBufferAlwaysFull ()Z
 	protected final fun isBufferEmpty ()Z
@@ -737,6 +746,8 @@ public class kotlinx/coroutines/experimental/channels/ConflatedChannel : kotlinx
 
 public class kotlinx/coroutines/experimental/channels/LinkedListChannel : kotlinx/coroutines/experimental/channels/AbstractChannel {
 	public fun <init> ()V
+	public fun <init> (Lkotlinx/coroutines/experimental/Job;)V
+	public synthetic fun <init> (Lkotlinx/coroutines/experimental/Job;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	protected final fun isBufferAlwaysEmpty ()Z
 	protected final fun isBufferAlwaysFull ()Z
 	protected final fun isBufferEmpty ()Z
@@ -775,6 +786,7 @@ public abstract interface class kotlinx/coroutines/experimental/channels/Produce
 
 public abstract interface class kotlinx/coroutines/experimental/channels/ReceiveChannel {
 	public abstract fun cancel (Ljava/lang/Throwable;)Z
+	public abstract fun getJob ()Lkotlinx/coroutines/experimental/Job;
 	public abstract fun getOnReceive ()Lkotlinx/coroutines/experimental/selects/SelectClause1;
 	public abstract fun getOnReceiveOrNull ()Lkotlinx/coroutines/experimental/selects/SelectClause1;
 	public abstract fun isClosedForReceive ()Z
@@ -797,6 +809,8 @@ public abstract interface class kotlinx/coroutines/experimental/channels/Receive
 
 public class kotlinx/coroutines/experimental/channels/RendezvousChannel : kotlinx/coroutines/experimental/channels/AbstractChannel {
 	public fun <init> ()V
+	public fun <init> (Lkotlinx/coroutines/experimental/Job;)V
+	public synthetic fun <init> (Lkotlinx/coroutines/experimental/Job;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	protected final fun isBufferAlwaysEmpty ()Z
 	protected final fun isBufferAlwaysFull ()Z
 	protected final fun isBufferEmpty ()Z
@@ -812,6 +826,7 @@ public abstract interface class kotlinx/coroutines/experimental/channels/Send {
 
 public abstract interface class kotlinx/coroutines/experimental/channels/SendChannel {
 	public abstract fun close (Ljava/lang/Throwable;)Z
+	public abstract fun getJob ()Lkotlinx/coroutines/experimental/Job;
 	public abstract fun getOnSend ()Lkotlinx/coroutines/experimental/selects/SelectClause2;
 	public abstract fun isClosedForSend ()Z
 	public abstract fun isFull ()Z

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/channels/AbstractChannel.kt

@@ -259,6 +259,8 @@ public abstract class AbstractSendChannel<E> : SendChannel<E> {
         helpClose(closed)
         onClosed(closed)
         afterClose(cause)
+        // Cancel it as the last action so if the channel is closed, then the job is cancelled as well
+        job.cancel(cause)
         return true
     }
 
@@ -473,9 +475,13 @@ public abstract class AbstractSendChannel<E> : SendChannel<E> {
 /**
  * Abstract send/receive channel. It is a base class for all channel implementations.
  */
-public abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E> {
+public abstract class AbstractChannel<E>(final override val job: Job) : AbstractSendChannel<E>(), Channel<E> {
     // ------ extension points for buffered channels ------
 
+    init {
+        registerCancellation(job)
+    }
+
     /**
      * Returns `true` if [isBufferEmpty] is always `true`.
      * @suppress **This is unstable API and it is subject to change.**
@@ -1051,3 +1057,20 @@ private abstract class Receive<in E> : LockFreeLinkedListNode(), ReceiveOrClosed
     abstract fun resumeReceiveClosed(closed: Closed<*>)
 }
 
+internal fun SendChannel<*>.registerCancellation(job: Job) {
+    val cancellation = ChannelCancellation(this, job)
+    job.invokeOnCompletion(cancellation.asHandler)
+}
+
+private class ChannelCancellation(
+    private val channel: SendChannel<*>, job: Job) : JobNode<Job>(job) {
+
+    override fun invoke(cause: Throwable?) {
+        if (job.isCancelled) {
+            channel.close(cause)
+        } else {
+            channel.close()
+        }
+    }
+}
+

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/channels/ArrayBroadcastChannel.kt

@@ -16,6 +16,7 @@
 
 package kotlinx.coroutines.experimental.channels
 
+import kotlinx.coroutines.experimental.*
 import kotlinx.coroutines.experimental.internal.*
 import kotlinx.coroutines.experimental.internalAnnotations.*
 import kotlinx.coroutines.experimental.selects.*
@@ -38,10 +39,16 @@ class ArrayBroadcastChannel<E>(
     /**
      * Buffer capacity.
      */
-    val capacity: Int
+    val capacity: Int,
+
+    /**
+     * Job owning this channel.
+     */
+    override val job: Job = Job()
 ) : AbstractSendChannel<E>(), BroadcastChannel<E> {
     init {
         require(capacity >= 1) { "ArrayBroadcastChannel capacity must be at least 1, but $capacity was specified" }
+        registerCancellation(job)
     }
 
     private val bufferLock = ReentrantLock()
@@ -195,7 +202,7 @@ class ArrayBroadcastChannel<E>(
 
     private class Subscriber<E>(
         private val broadcastChannel: ArrayBroadcastChannel<E>
-    ) : AbstractChannel<E>(), SubscriptionReceiveChannel<E> {
+    ) : AbstractChannel<E>(Job()), SubscriptionReceiveChannel<E> {
         private val subLock = ReentrantLock()
 
         @Volatile

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/channels/ArrayChannel.kt

@@ -16,8 +16,9 @@
 
 package kotlinx.coroutines.experimental.channels
 
+import kotlinx.coroutines.experimental.*
 import kotlinx.coroutines.experimental.internal.*
-import kotlinx.coroutines.experimental.internalAnnotations.Volatile
+import kotlinx.coroutines.experimental.internalAnnotations.*
 import kotlinx.coroutines.experimental.selects.*
 
 /**
@@ -33,8 +34,13 @@ public open class ArrayChannel<E>(
     /**
      * Buffer capacity.
      */
-    val capacity: Int
-) : AbstractChannel<E>() {
+    val capacity: Int,
+
+    /**
+     * Job owning this channel.
+     */
+    job: Job = Job()
+) : AbstractChannel<E>(job) {
     init {
         require(capacity >= 1) { "ArrayChannel capacity must be at least 1, but $capacity was specified" }
     }

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/channels/Channel.kt

@@ -16,20 +16,24 @@
 
 package kotlinx.coroutines.experimental.channels
 
-import kotlinx.coroutines.experimental.CancellationException
-import kotlinx.coroutines.experimental.CoroutineScope
-import kotlinx.coroutines.experimental.Job
+import kotlinx.coroutines.experimental.*
 import kotlinx.coroutines.experimental.channels.Channel.Factory.CONFLATED
 import kotlinx.coroutines.experimental.channels.Channel.Factory.UNLIMITED
-import kotlinx.coroutines.experimental.selects.SelectClause1
-import kotlinx.coroutines.experimental.selects.SelectClause2
-import kotlinx.coroutines.experimental.selects.select
-import kotlinx.coroutines.experimental.yield
+import kotlinx.coroutines.experimental.selects.*
 
 /**
  * Sender's interface to [Channel].
  */
 public interface SendChannel<in E> {
+
+    /**
+     * The job of this channel bounded with channel lifecycle.
+     * If job is completed with any reason (either normally or exceptionally), channel is [closed][SendChannel.close]
+     * with a completion [cause][Job.getCancellationException] of the job.
+     * If the channel is [closed][isClosedForSend], job is cancelled with the same reason as [SendChannel.close] call
+     */
+    public val job: Job
+
     /**
      * Returns `true` if this channel was closed by invocation of [close] and thus
      * the [send] and [offer] attempts throws exception.
@@ -105,6 +109,15 @@ public interface SendChannel<in E> {
  * Receiver's interface to [Channel].
  */
 public interface ReceiveChannel<out E> {
+
+    /**
+     * The job of this channel bounded with channel lifecycle.
+     * If job is completed with any reason (either normally or exceptionally), channel is [cancelled][ReceiveChannel.cancel]
+     * with a completion [cause][Job.getCancellationException] of the job.
+     * If the channel is cancelled or [closed][isClosedForReceive], job is cancelled with the same reason as [ReceiveChannel.cancel] call
+     */
+    public val job: Job
+
     /**
      * Returns `true` if this channel was closed by invocation of [close][SendChannel.close] on the [SendChannel]
      * side and all previously sent items were already received, so that the [receive] attempt

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/channels/ConflatedBroadcastChannel.kt

@@ -17,6 +17,7 @@
 package kotlinx.coroutines.experimental.channels
 
 import kotlinx.atomicfu.*
+import kotlinx.coroutines.experimental.*
 import kotlinx.coroutines.experimental.internal.*
 import kotlinx.coroutines.experimental.internalAnnotations.*
 import kotlinx.coroutines.experimental.intrinsics.*
@@ -37,7 +38,12 @@ import kotlinx.coroutines.experimental.selects.*
  * [opening][openSubscription] and [closing][SubscriptionReceiveChannel.close] subscription takes O(N) time, where N is the
  * number of subscribers.
  */
-public class ConflatedBroadcastChannel<E>() : BroadcastChannel<E> {
+public class ConflatedBroadcastChannel<E>(override val job: Job = Job()) : BroadcastChannel<E> {
+
+    init {
+        registerCancellation(job)
+    }
+
     /**
      * Creates an instance of this class that already holds a value.
      *

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/channels/ConflatedChannel.kt

@@ -16,8 +16,8 @@
 
 package kotlinx.coroutines.experimental.channels
 
-import kotlinx.coroutines.experimental.selects.ALREADY_SELECTED
-import kotlinx.coroutines.experimental.selects.SelectInstance
+import kotlinx.coroutines.experimental.*
+import kotlinx.coroutines.experimental.selects.*
 
 /**
  * Channel that buffers at most one element and conflates all subsequent `send` and `offer` invocations,
@@ -30,7 +30,7 @@ import kotlinx.coroutines.experimental.selects.SelectInstance
  *
  * This implementation is fully lock-free.
  */
-public open class ConflatedChannel<E> : AbstractChannel<E>() {
+public open class ConflatedChannel<E>(job: Job = Job()) : AbstractChannel<E>(job) {
     protected final override val isBufferAlwaysEmpty: Boolean get() = true
     protected final override val isBufferEmpty: Boolean get() = true
     protected final override val isBufferAlwaysFull: Boolean get() = false

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/channels/LinkedListChannel.kt

@@ -16,8 +16,8 @@
 
 package kotlinx.coroutines.experimental.channels
 
-import kotlinx.coroutines.experimental.selects.ALREADY_SELECTED
-import kotlinx.coroutines.experimental.selects.SelectInstance
+import kotlinx.coroutines.experimental.*
+import kotlinx.coroutines.experimental.selects.*
 
 /**
  * Channel with linked-list buffer of a unlimited capacity (limited only by available memory).
@@ -27,7 +27,7 @@ import kotlinx.coroutines.experimental.selects.SelectInstance
  *
  * This implementation is fully lock-free.
  */
-public open class LinkedListChannel<E> : AbstractChannel<E>() {
+public open class LinkedListChannel<E>(job: Job = Job()) : AbstractChannel<E>(job) {
     protected final override val isBufferAlwaysEmpty: Boolean get() = true
     protected final override val isBufferEmpty: Boolean get() = true
     protected final override val isBufferAlwaysFull: Boolean get() = false

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/channels/RendezvousChannel.kt

@@ -16,6 +16,8 @@
 
 package kotlinx.coroutines.experimental.channels
 
+import kotlinx.coroutines.experimental.*
+
 /**
  * Rendezvous channel. This channel does not have any buffer at all. An element is transferred from sender
  * to receiver only when [send] and [receive] invocations meet in time (rendezvous), so [send] suspends
@@ -25,7 +27,7 @@ package kotlinx.coroutines.experimental.channels
  *
  * This implementation is fully lock-free.
  */
-public open class RendezvousChannel<E> : AbstractChannel<E>() {
+public open class RendezvousChannel<E>(job: Job = Job()) : AbstractChannel<E>(job) {
     protected final override val isBufferAlwaysEmpty: Boolean get() = true
     protected final override val isBufferEmpty: Boolean get() = true
     protected final override val isBufferAlwaysFull: Boolean get() = true

common/kotlinx-coroutines-core-common/src/test/kotlin/kotlinx/coroutines/experimental/TestBase.common.kt

@@ -16,6 +16,8 @@
 
 package kotlinx.coroutines.experimental
 
+import kotlin.test.*
+
 public expect open class TestBase constructor() {
     public val isStressTest: Boolean
     public val stressTestMultiplier: Int
@@ -31,3 +33,23 @@ public expect open class TestBase constructor() {
         block: suspend CoroutineScope.() -> Unit
     )
 }
+
+suspend inline fun <reified T : Exception> assertFailsWith(deferred: Deferred<*>) {
+    try {
+        deferred.await()
+        fail("Expected ${T::class} to be thrown, but was completed successfully")
+    } catch (e: Exception) {
+        assertTrue(e is T, "Expected exception of type ${T::class}, but has $e}")
+    }
+}
+
+// Clashes with assertFailsWith
+suspend inline fun <reified T : Throwable> assertFails(noinline block: suspend () -> Unit): T {
+    try {
+        block()
+        fail("Expected ${T::class} to be thrown, but was completed successfully")
+    } catch (e: Throwable) {
+        assertTrue(e is T, "Expected exception of type ${T::class}, but has $e}")
+        return e as T
+    }
+}