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 #127
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/ArrayChannel.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/ArrayChannel.kt
index 4d2c5c3..8ce9008 100644
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/ArrayChannel.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/ArrayChannel.kt
@@ -231,4 +231,18 @@
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()
+ }
}
\ No newline at end of file