Introduced ReceiveChannel.cancel method;

all operators on ReceiveChannel fully consume the original channel
using a helper consume extension, which is reflected in docs;
removed `suspend` modifier from intermediate channel operators;
consistently renamed channel type param to <E>;
added two versions for xxxTo fun -- with MutableList & SendChannel;
added tests for all channel operators;
dropped/deprecated ActorJob/ProducerJob, fixes Kotlin#127

core/kotlinx-coroutines-core/README.md

@@ -8,8 +8,8 @@ Coroutine builder functions:
 | ------------- | ------------- | ---------------- | ---------------
 | [launch] | [Job] | [CoroutineScope] | Launches coroutine that does not have any result 
 | [async] | [Deferred] | [CoroutineScope] | Returns a single value with the future result
-| [produce][kotlinx.coroutines.experimental.channels.produce] | [ProducerJob][kotlinx.coroutines.experimental.channels.ProducerJob] | [ProducerScope][kotlinx.coroutines.experimental.channels.ProducerScope] | Produces a stream of elements
-| [actor][kotlinx.coroutines.experimental.channels.actor] | [ActorJob][kotlinx.coroutines.experimental.channels.ActorJob] | [ActorScope][kotlinx.coroutines.experimental.channels.ActorScope] | Processes a stream of messages
+| [produce][kotlinx.coroutines.experimental.channels.produce] | [ReceiveChannel][kotlinx.coroutines.experimental.channels.ReceiveChannel] | [ProducerScope][kotlinx.coroutines.experimental.channels.ProducerScope] | Produces a stream of elements
+| [actor][kotlinx.coroutines.experimental.channels.actor] | [SendChannel][kotlinx.coroutines.experimental.channels.SendChannel] | [ActorScope][kotlinx.coroutines.experimental.channels.ActorScope] | Processes a stream of messages
 | [runBlocking] | `T` | [CoroutineScope] | Blocks the thread while the coroutine runs
 
 Coroutine dispatchers implementing [CoroutineDispatcher]:

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt

@@ -898,7 +898,7 @@ public open class JobSupport(active: Boolean) : Job, SelectClause0, SelectClause
 
  protected open val hasCancellingState: Boolean get() = false
 
- public final override fun cancel(cause: Throwable?): Boolean =
+ public override fun cancel(cause: Throwable?): Boolean =
  if (hasCancellingState)
  makeCancelling(cause) else
  makeCancelled(cause)

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

@@ -398,6 +398,11 @@ public abstract class AbstractSendChannel<E> : SendChannel<E> {
  remove()
  }
 
+ override fun resumeSendClosed(closed: Closed<*>) {
+ if (select.trySelect(null))
+ select.resumeSelectCancellableWithException(closed.sendException)
+ }
+
  override fun toString(): String = "SendSelect($pollResult)[$channel, $select]"
  }
 
@@ -407,6 +412,7 @@ public abstract class AbstractSendChannel<E> : SendChannel<E> {
  override val pollResult: Any? get() = element
  override fun tryResumeSend(idempotent: Any?): Any? = SEND_RESUMED
  override fun completeResumeSend(token: Any) { check(token === SEND_RESUMED) }
+ override fun resumeSendClosed(closed: Closed<*>) {}
  }
 }
 
@@ -567,6 +573,24 @@ public abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E>
  return if (result === POLL_FAILED) null else receiveOrNullResult(result)
  }
 
+ override fun cancel(cause: Throwable?): Boolean =
+ close(cause).also {
+ cleanupSendQueueOnCancel()
+ }
+
+ // Note: this function is invoked when channel is already closed
+ protected open fun cleanupSendQueueOnCancel() {
+ val closed = closedForSend ?: error("Cannot happen")
+ while (true) {
+ val send = takeFirstSendOrPeekClosed() ?: error("Cannot happen")
+ if (send is Closed<*>) {
+ check(send === closed)
+ return // cleaned
+ }
+ send.resumeSendClosed(closed)
+ }
+ }
+
  public final override fun iterator(): ChannelIterator<E> = Itr(this)
 
  // ------ registerSelectReceive ------
@@ -909,6 +933,7 @@ public interface Send {
  val pollResult: Any? // E | Closed
  fun tryResumeSend(idempotent: Any?): Any?
  fun completeResumeSend(token: Any)
+ fun resumeSendClosed(closed: Closed<*>)
 }
 
 /**
@@ -922,7 +947,7 @@ public interface ReceiveOrClosed<in E> {
 }
 
 /**
- * Represents closed channel.
+ * Represents sender for a specific element.
  * @suppress **This is unstable API and it is subject to change.**
  */
 @Suppress("UNCHECKED_CAST")
@@ -932,6 +957,7 @@ public class SendElement(
 ) : LockFreeLinkedListNode(), Send {
  override fun tryResumeSend(idempotent: Any?): Any? = cont.tryResume(Unit, idempotent)
  override fun completeResumeSend(token: Any) = cont.completeResume(token)
+ override fun resumeSendClosed(closed: Closed<*>) = cont.resumeWithException(closed.sendException)
  override fun toString(): String = "SendElement($pollResult)[$cont]"
 }
 
@@ -951,6 +977,7 @@ public class Closed<in E>(
  override fun completeResumeSend(token: Any) { check(token === CLOSE_RESUMED) }
  override fun tryResumeReceive(value: E, idempotent: Any?): Any? = CLOSE_RESUMED
  override fun completeResumeReceive(token: Any) { check(token === CLOSE_RESUMED) }
+ override fun resumeSendClosed(closed: Closed<*>) = error("Should be never invoked")
  override fun toString(): String = "Closed[$closeCause]"
 }
 

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

@@ -37,14 +37,11 @@ public interface ActorScope<E> : CoroutineScope, ReceiveChannel<E> {
 }
 
 /**
- * Return type for [actor] coroutine builder.
+ * @suppress **Deprecated**: Use `SendChannel`.
  */
-public interface ActorJob<in E> : Job, SendChannel<E> {
- /**
- * A reference to the mailbox channel that this coroutine is receiving messages from.
- * All the [SendChannel] functions on this interface delegate to
- * the channel instance returned by this function.
- */
+@Deprecated(message = "Use `SendChannel`", replaceWith = ReplaceWith("SendChannel"))
+interface ActorJob<in E> : SendChannel<E> {
+ @Deprecated(message = "Use SendChannel itself")
  val channel: SendChannel<E>
 }
 
@@ -87,7 +84,7 @@ public fun <E> actor(
  capacity: Int = 0,
  start: CoroutineStart = CoroutineStart.DEFAULT,
  block: suspend ActorScope<E>.() -> Unit
-): ActorJob<E> {
+): SendChannel<E> {
  val newContext = newCoroutineContext(context)
  val channel = Channel<E>(capacity)
  val coroutine = if (start.isLazy)
@@ -109,8 +106,6 @@ private class LazyActorCoroutine<E>(
  channel: Channel<E>,
  private val block: suspend ActorScope<E>.() -> Unit
 ) : ActorCoroutine<E>(parentContext, channel, active = false), SelectClause2<E, SendChannel<E>> {
- override val channel: Channel<E> get() = this
-
  override fun onStart() {
  block.startCoroutineCancellable(this, this)
  }

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

@@ -209,10 +209,10 @@ class ArrayBroadcastChannel<E>(
  override val isBufferAlwaysFull: Boolean get() = error("Should not be used")
  override val isBufferFull: Boolean get() = error("Should not be used")
 
- override fun close() {
- if (close(cause = null))
- broadcastChannel.updateHead(removeSub = this)
- }
+ override fun cancel(cause: Throwable?): Boolean =
+ close(cause).also { closed ->
+ if (closed) broadcastChannel.updateHead(removeSub = this)
+  }
 
  // returns true if subHead was updated and broadcast channel's head must be checked
  // this method is lock-free (it never waits on lock)

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

@@ -231,4 +231,18 @@ public open class ArrayChannel<E>(
  send!!.completeResumeSend(token!!)
  return result
  }
+
+ // Note: this function is invoked when channel is already closed
+ override fun cleanupSendQueueOnCancel() {
+ // clear buffer first
+ lock.withLock {
+ repeat(size) {
+ buffer[head] = 0
+ head = (head + 1) % capacity
+ }
+ size = 0
+ }
+ // then clean all queued senders
+ super.cleanupSendQueueOnCancel()
+ }
 }

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

@@ -76,10 +76,12 @@ public fun <E> BroadcastChannel(capacity: Int): BroadcastChannel<E> =
 /**
  * Return type for [BroadcastChannel.openSubscription] that can be used to [receive] elements from the
  * open subscription and to [close] it to unsubscribe.
+ *
+ * Note, that invocation of [cancel] also closes subscription.
  */
 public interface SubscriptionReceiveChannel<out T> : ReceiveChannel<T>, Closeable {
  /**
- * Closes this subscription.
+ * Closes this subscription. This is a synonym for [cancel].
  */
- public override fun close()
+ public override fun close() { cancel() }
 }

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

@@ -87,7 +87,9 @@ public interface SendChannel<in E> {
  /**
  * Closes this channel with an optional exceptional [cause].
  * This is an idempotent operation -- repeated invocations of this function have no effect and return `false`.
- * Conceptually, its sends a special "close token" over this channel. Immediately after invocation of this function
+ * Conceptually, its sends a special "close token" over this channel.
+ *
+ * Immediately after invocation of this function
  * [isClosedForSend] starts returning `true`. However, [isClosedForReceive][ReceiveChannel.isClosedForReceive]
  * on the side of [ReceiveChannel] starts returning `true` only after all previously sent elements
  * are received.
@@ -192,6 +194,21 @@ public interface ReceiveChannel<out E> {
  * throws the original [close][SendChannel.close] cause exception if the channel has _failed_.
  */
  public operator fun iterator(): ChannelIterator<E>
+
+ /**
+ * Cancels reception of remaining elements from this channel. This function closes the channel with
+ * the specified cause (unless it was already closed) and removes all buffered sent elements from it.
+ * This function returns `true` if the channel was not closed previously, or `false` otherwise.
+ *
+ * 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] if it was cancelled without a cause.
+ * A channel that was cancelled with non-null [cause] is called a _failed_ channel. Attempts to send or
+ * receive on a failed channel throw the specified [cause] exception.
+ */
+ public fun cancel(cause: Throwable? = null): Boolean
 }
 
 /**

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

@@ -17,18 +17,22 @@
 package kotlinx.coroutines.experimental.channels
 
 import kotlinx.coroutines.experimental.AbstractCoroutine
-import kotlinx.coroutines.experimental.JobSupport
 import kotlinx.coroutines.experimental.handleCoroutineException
 import kotlin.coroutines.experimental.CoroutineContext
 
 internal open class ChannelCoroutine<E>(
  parentContext: CoroutineContext,
- open val channel: Channel<E>,
+ channel: Channel<E>,
  active: Boolean
 ) : AbstractCoroutine<Unit>(parentContext, active), Channel<E> by channel {
- override fun afterCompletion(state: Any?, mode: Int) {
- val cause = (state as? JobSupport.CompletedExceptionally)?.cause
- if (!channel.close(cause) && cause != null)
+ val channel: Channel<E>
+ get() = this
+
+ override fun onCancellation(exceptionally: CompletedExceptionally?) {
+ val cause = exceptionally?.cause
+ if (!close(cause) && cause != null)
  handleCoroutineException(context, cause)
  }
+
+ override fun cancel(cause: Throwable?): Boolean = super.cancel(cause)
 }