consumeAll renamed to cancel; cleaned up some channel operators;

dropped/deprecated ActorJob/ProducerJob, fixes #127

Author: elizarov

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

@@ -573,14 +573,13 @@ public abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E>
         return if (result === POLL_FAILED) null else receiveOrNullResult(result)
     }
 
-    override fun consumeAll() {
-        close(cause = null)
-        // now cleanup send queue
-        consumeAllInternal()
-    }
+    override fun cancel(cause: Throwable?): Boolean =
+        close(cause).also {
+            cleanupSendQueueOnCancel()
+        }
 
     // Note: this function is invoked when channel is already closed
-    protected open fun consumeAllInternal() {
+    protected open fun cleanupSendQueueOnCancel() {
         val closed = closedForSend ?: error("Cannot happen")
         while (true) {
             val send = takeFirstSendOrPeekClosed() ?: error("Cannot happen")

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 consumeAll() {
-            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

@@ -233,7 +233,7 @@ public open class ArrayChannel<E>(
     }
 
     // Note: this function is invoked when channel is already closed
-    override fun consumeAllInternal() {
+    override fun cleanupSendQueueOnCancel() {
         // clear buffer first
         lock.withLock {
             repeat(size) {
@@ -243,6 +243,6 @@ public open class ArrayChannel<E>(
             size = 0
         }
         // then clean all queued senders
-        super.consumeAllInternal()
+        super.cleanupSendQueueOnCancel()
     }
 }

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

@@ -77,11 +77,11 @@ 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 [consumeAll] also closes subscription.
+ * Note, that invocation of [cancel] also closes subscription.
  */
 public interface SubscriptionReceiveChannel<out T> : ReceiveChannel<T>, Closeable {
     /**
-     * Closes this subscription. This is a synonym for [consumeAll].
+     * Closes this subscription. This is a synonym for [cancel].
      */
-    public override fun close() = consumeAll()
+    public override fun close() { cancel() }
 }

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

@@ -196,16 +196,19 @@ public interface ReceiveChannel<out E> {
     public operator fun iterator(): ChannelIterator<E>
 
     /**
-     * Consumes all remaining elements from this channel. This function marks the channel as _consumed_
-     * by closing it normally (unless it was already closed) and removes all buffered sent elements from it.
+     * 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].
+     * [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 consumeAll()
+    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
+    val channel: Channel<E>
+        get() = this
+
+    override fun onCancellation(exceptionally: CompletedExceptionally?) {
+        val cause = exceptionally?.cause
         if (!channel.close(cause) && cause != null)
             handleCoroutineException(context, cause)
     }
+
+    override fun cancel(cause: Throwable?): Boolean = super.cancel(cause)
 }