Improve ReceiveChannel operators implementations to guarantee closing
of the source channels under all circumstances;
`onCompletion` added to `produce` builder;
`ReceiveChannel.consumes(): CompletionHandler` extension fun.
Fixes #279
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Channels.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Channels.kt
index 4623ece..87abbf5 100644
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Channels.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Channels.kt
@@ -78,7 +78,7 @@
/**
* Subscribes to this [BroadcastChannel] and performs the specified action for each received element.
*/
-public inline suspend fun <E> BroadcastChannel<E>.consumeEach(action: (E) -> Unit) =
+public suspend inline fun <E> BroadcastChannel<E>.consumeEach(action: (E) -> Unit) =
consume {
for (element in this) action(element)
}
@@ -93,24 +93,78 @@
// -------- Operations on ReceiveChannel --------
/**
+ * Returns a [CompletionHandler] that invokes [cancel][ReceiveChannel.cancel] on the [ReceiveChannel]
+ * with the corresponding cause. See also [ReceiveChannel.consume].
+ *
+ * **WARNING**: It is planned that in the future a second invocation of this method
+ * on an channel that is already being consumed is going to fail fast, that is
+ * immediately throw an [IllegalStateException].
+ * See [this issue](https://github.com/Kotlin/kotlinx.coroutines/issues/167)
+ * for details.
+ */
+public fun ReceiveChannel<*>.consumes(): CompletionHandler =
+ { cause: Throwable? -> cancel(cause) }
+
+/**
+ * Returns a [CompletionHandler] that invokes [cancel][ReceiveChannel.cancel] on all the
+ * specified [ReceiveChannel] instances with the corresponding cause.
+ * See also [ReceiveChannel.consumes()] for a version on one channel.
+ */
+public fun consumesAll(vararg channels: ReceiveChannel<*>): CompletionHandler =
+ { cause: Throwable? ->
+ var exception: Throwable? = null
+ for (channel in channels)
+ try {
+ channel.cancel(cause)
+ } catch (e: Throwable) {
+ if (exception == null) {
+ exception = e
+ } else {
+ exception.addSuppressed(e)
+ }
+ }
+ exception?.let { throw it }
+ }
+
+/**
* Makes sure that the given [block] consumes all elements from the given channel
* by always invoking [cancel][ReceiveChannel.cancel] after the execution of the block.
+ *
+ * **WARNING**: It is planned that in the future a second invocation of this method
+ * on an channel that is already being consumed is going to fail fast, that is
+ * immediately throw an [IllegalStateException].
+ * See [this issue](https://github.com/Kotlin/kotlinx.coroutines/issues/167)
+ * for details.
+ *
+ * The operation is _terminal_.
*/
-public inline fun <E, R> ReceiveChannel<E>.consume(block: ReceiveChannel<E>.() -> R): R =
+public inline fun <E, R> ReceiveChannel<E>.consume(block: ReceiveChannel<E>.() -> R): R {
+ var cause: Throwable? = null
try {
- block()
+ return block()
+ } catch (e: Throwable) {
+ cause = e
+ throw e
} finally {
- cancel()
+ cancel(cause)
}
+}
/**
* Performs the given [action] for each received element.
*
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * **WARNING**: It is planned that in the future a second invocation of this method
+ * on an channel that is already being consumed is going to fail fast, that is
+ * immediately throw an [IllegalStateException].
+ * See [this issue](https://github.com/Kotlin/kotlinx.coroutines/issues/167)
+ * for details.
+ *
+ * The operation is _terminal_.
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.consumeEach(action: (E) -> Unit) =
+public suspend inline fun <E> ReceiveChannel<E>.consumeEach(action: (E) -> Unit) =
consume {
- for (element in this) action(element)
+ for (e in this) action(e)
}
/**
@@ -123,9 +177,10 @@
/**
* Performs the given [action] for each received element.
*
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * The operation is _terminal_.
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.consumeEachIndexed(action: (IndexedValue<E>) -> Unit) {
+public suspend inline fun <E> ReceiveChannel<E>.consumeEachIndexed(action: (IndexedValue<E>) -> Unit) {
var index = 0
consumeEach {
action(IndexedValue(index++, it))
@@ -136,7 +191,7 @@
* Returns an element at the given [index] or throws an [IndexOutOfBoundsException] if the [index] is out of bounds of this channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.elementAt(index: Int): E =
elementAtOrElse(index) { throw IndexOutOfBoundsException("ReceiveChannel doesn't contain element at index $index.") }
@@ -145,9 +200,9 @@
* Returns an element at the given [index] or the result of calling the [defaultValue] function if the [index] is out of bounds of this channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.elementAtOrElse(index: Int, defaultValue: (Int) -> E): E =
+public suspend inline fun <E> ReceiveChannel<E>.elementAtOrElse(index: Int, defaultValue: (Int) -> E): E =
consume {
if (index < 0)
return defaultValue(index)
@@ -163,7 +218,7 @@
* Returns an element at the given [index] or `null` if the [index] is out of bounds of this channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.elementAtOrNull(index: Int): E? =
consume {
@@ -181,18 +236,18 @@
* Returns the first element matching the given [predicate], or `null` if no such element was found.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.find(predicate: (E) -> Boolean): E? =
+public suspend inline fun <E> ReceiveChannel<E>.find(predicate: (E) -> Boolean): E? =
firstOrNull(predicate)
/**
* Returns the last element matching the given [predicate], or `null` if no such element was found.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.findLast(predicate: (E) -> Boolean): E? =
+public suspend inline fun <E> ReceiveChannel<E>.findLast(predicate: (E) -> Boolean): E? =
lastOrNull(predicate)
/**
@@ -200,7 +255,7 @@
* @throws [NoSuchElementException] if the channel is empty.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.first(): E =
consume {
@@ -215,9 +270,9 @@
* @throws [NoSuchElementException] if no such element is found.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.first(predicate: (E) -> Boolean): E {
+public suspend inline fun <E> ReceiveChannel<E>.first(predicate: (E) -> Boolean): E {
consumeEach {
if (predicate(it)) return it
}
@@ -228,7 +283,7 @@
* Returns the first element, or `null` if the channel is empty.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.firstOrNull(): E? =
consume {
@@ -242,7 +297,7 @@
* Returns the first element matching the given [predicate], or `null` if element was not found.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public inline suspend fun <E> ReceiveChannel<E>.firstOrNull(predicate: (E) -> Boolean): E? {
consumeEach {
@@ -255,7 +310,7 @@
* Returns first index of [element], or -1 if the channel does not contain element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.indexOf(element: E): Int {
var index = 0
@@ -271,9 +326,9 @@
* Returns index of the first element matching the given [predicate], or -1 if the channel does not contain such element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.indexOfFirst(predicate: (E) -> Boolean): Int {
+public suspend inline fun <E> ReceiveChannel<E>.indexOfFirst(predicate: (E) -> Boolean): Int {
var index = 0
consumeEach {
if (predicate(it))
@@ -287,7 +342,7 @@
* Returns index of the last element matching the given [predicate], or -1 if the channel does not contain such element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public inline suspend fun <E> ReceiveChannel<E>.indexOfLast(predicate: (E) -> Boolean): Int {
var lastIndex = -1
@@ -305,7 +360,7 @@
* @throws [NoSuchElementException] if the channel is empty.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.last(): E =
consume {
@@ -323,9 +378,9 @@
* @throws [NoSuchElementException] if no such element is found.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.last(predicate: (E) -> Boolean): E {
+public suspend inline fun <E> ReceiveChannel<E>.last(predicate: (E) -> Boolean): E {
var last: E? = null
var found = false
consumeEach {
@@ -343,7 +398,7 @@
* Returns last index of [element], or -1 if the channel does not contain element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.lastIndexOf(element: E): Int {
var lastIndex = -1
@@ -360,7 +415,7 @@
* Returns the last element, or `null` if the channel is empty.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.lastOrNull(): E? =
consume {
@@ -377,9 +432,9 @@
* Returns the last element matching the given [predicate], or `null` if no such element was found.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.lastOrNull(predicate: (E) -> Boolean): E? {
+public suspend inline fun <E> ReceiveChannel<E>.lastOrNull(predicate: (E) -> Boolean): E? {
var last: E? = null
consumeEach {
if (predicate(it)) {
@@ -393,7 +448,7 @@
* Returns the single element, or throws an exception if the channel is empty or has more than one element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.single(): E =
consume {
@@ -410,9 +465,9 @@
* Returns the single element matching the given [predicate], or throws exception if there is no or more than one matching element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.single(predicate: (E) -> Boolean): E {
+public suspend inline fun <E> ReceiveChannel<E>.single(predicate: (E) -> Boolean): E {
var single: E? = null
var found = false
consumeEach {
@@ -431,7 +486,7 @@
* Returns single element, or `null` if the channel is empty or has more than one element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.singleOrNull(): E? =
consume {
@@ -448,9 +503,9 @@
* Returns the single element matching the given [predicate], or `null` if element was not found or more than one element was found.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.singleOrNull(predicate: (E) -> Boolean): E? {
+public suspend inline fun <E> ReceiveChannel<E>.singleOrNull(predicate: (E) -> Boolean): E? {
var single: E? = null
var found = false
consumeEach {
@@ -468,22 +523,20 @@
* Returns a channel containing all elements except first [n] elements.
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public fun <E> ReceiveChannel<E>.drop(n: Int, context: CoroutineContext = Unconfined): ReceiveChannel<E> =
- produce(context) {
- consume {
- require(n >= 0) { "Requested element count $n is less than zero." }
- var remaining: Int = n
- if (remaining > 0)
- for (element in this@drop) {
- remaining--
- if (remaining == 0)
- break
- }
- for (element in this@drop) {
- send(element)
+ produce(context, onCompletion = consumes()) {
+ require(n >= 0) { "Requested element count $n is less than zero." }
+ var remaining: Int = n
+ if (remaining > 0)
+ for (e in this@drop) {
+ remaining--
+ if (remaining == 0)
+ break
}
+ for (e in this@drop) {
+ send(e)
}
}
@@ -491,21 +544,19 @@
* Returns a channel containing all elements except first elements that satisfy the given [predicate].
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E> ReceiveChannel<E>.dropWhile(context: CoroutineContext = Unconfined, predicate: suspend (E) -> Boolean): ReceiveChannel<E> =
- produce(context) {
- consume {
- for (element in this@dropWhile) {
- if (!predicate(element)) {
- send(element)
- break
- }
+ produce(context, onCompletion = consumes()) {
+ for (e in this@dropWhile) {
+ if (!predicate(e)) {
+ send(e)
+ break
}
- for (element in this@dropWhile) {
- send(element)
- }
+ }
+ for (e in this@dropWhile) {
+ send(e)
}
}
@@ -513,13 +564,13 @@
* Returns a channel containing only elements matching the given [predicate].
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E> ReceiveChannel<E>.filter(context: CoroutineContext = Unconfined, predicate: suspend (E) -> Boolean): ReceiveChannel<E> =
- produce(context) {
- consumeEach {
- if (predicate(it)) send(it)
+ produce(context, onCompletion = consumes()) {
+ for (e in this@filter) {
+ if (predicate(e)) send(e)
}
}
@@ -529,14 +580,14 @@
* and returns the result of predicate evaluation on the element.
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E> ReceiveChannel<E>.filterIndexed(context: CoroutineContext = Unconfined, predicate: suspend (index: Int, E) -> Boolean): ReceiveChannel<E> =
- produce(context) {
+ produce(context, onCompletion = consumes()) {
var index = 0
- consumeEach {
- if (predicate(index++, it)) send(it)
+ for (e in this@filterIndexed) {
+ if (predicate(index++, e)) send(e)
}
}
@@ -546,9 +597,9 @@
* and returns the result of predicate evaluation on the element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, C : MutableCollection<in E>> ReceiveChannel<E>.filterIndexedTo(destination: C, predicate: (index: Int, E) -> Boolean): C {
+public suspend inline fun <E, C : MutableCollection<in E>> ReceiveChannel<E>.filterIndexedTo(destination: C, predicate: (index: Int, E) -> Boolean): C {
consumeEachIndexed { (index, element) ->
if (predicate(index, element)) destination.add(element)
}
@@ -561,9 +612,9 @@
* and returns the result of predicate evaluation on the element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, C : SendChannel<E>> ReceiveChannel<E>.filterIndexedTo(destination: C, predicate: (index: Int, E) -> Boolean): C {
+public suspend inline fun <E, C : SendChannel<E>> ReceiveChannel<E>.filterIndexedTo(destination: C, predicate: (index: Int, E) -> Boolean): C {
consumeEachIndexed { (index, element) ->
if (predicate(index, element)) destination.send(element)
}
@@ -574,18 +625,13 @@
* Returns a channel containing all elements not matching the given [predicate].
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E> ReceiveChannel<E>.filterNot(context: CoroutineContext = Unconfined, predicate: suspend (E) -> Boolean): ReceiveChannel<E> =
filter(context) { !predicate(it) }
/**
- * Returns a channel containing all elements not matching the given [predicate].
- *
- * The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
- *
* @suppress **Deprecated**: For binary compatibility only
*/
@Deprecated("For binary compatibility only", level = DeprecationLevel.HIDDEN)
@@ -595,7 +641,7 @@
* Returns a channel containing all elements that are not `null`.
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
@Suppress("UNCHECKED_CAST")
public fun <E : Any> ReceiveChannel<E?>.filterNotNull(): ReceiveChannel<E> =
@@ -605,7 +651,7 @@
* Appends all elements that are not `null` to the given [destination].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E : Any, C : MutableCollection<in E>> ReceiveChannel<E?>.filterNotNullTo(destination: C): C {
consumeEach {
@@ -618,7 +664,7 @@
* Appends all elements that are not `null` to the given [destination].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E : Any, C : SendChannel<E>> ReceiveChannel<E?>.filterNotNullTo(destination: C): C {
consumeEach {
@@ -631,9 +677,9 @@
* Appends all elements not matching the given [predicate] to the given [destination].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, C : MutableCollection<in E>> ReceiveChannel<E>.filterNotTo(destination: C, predicate: (E) -> Boolean): C {
+public suspend inline fun <E, C : MutableCollection<in E>> ReceiveChannel<E>.filterNotTo(destination: C, predicate: (E) -> Boolean): C {
consumeEach {
if (!predicate(it)) destination.add(it)
}
@@ -644,9 +690,9 @@
* Appends all elements not matching the given [predicate] to the given [destination].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, C : SendChannel<E>> ReceiveChannel<E>.filterNotTo(destination: C, predicate: (E) -> Boolean): C {
+public suspend inline fun <E, C : SendChannel<E>> ReceiveChannel<E>.filterNotTo(destination: C, predicate: (E) -> Boolean): C {
consumeEach {
if (!predicate(it)) destination.send(it)
}
@@ -657,9 +703,9 @@
* Appends all elements matching the given [predicate] to the given [destination].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, C : MutableCollection<in E>> ReceiveChannel<E>.filterTo(destination: C, predicate: (E) -> Boolean): C {
+public suspend inline fun <E, C : MutableCollection<in E>> ReceiveChannel<E>.filterTo(destination: C, predicate: (E) -> Boolean): C {
consumeEach {
if (predicate(it)) destination.add(it)
}
@@ -670,9 +716,9 @@
* Appends all elements matching the given [predicate] to the given [destination].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, C : SendChannel<E>> ReceiveChannel<E>.filterTo(destination: C, predicate: (E) -> Boolean): C {
+public suspend inline fun <E, C : SendChannel<E>> ReceiveChannel<E>.filterTo(destination: C, predicate: (E) -> Boolean): C {
consumeEach {
if (predicate(it)) destination.send(it)
}
@@ -683,20 +729,18 @@
* Returns a channel containing first [n] elements.
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public fun <E> ReceiveChannel<E>.take(n: Int, context: CoroutineContext = Unconfined): ReceiveChannel<E> =
- produce(context) {
- consume {
- if (n == 0) return@produce
- require(n >= 0) { "Requested element count $n is less than zero." }
- var remaining: Int = n
- for (element in this@take) {
- send(element)
- remaining--
- if (remaining == 0)
- return@produce
- }
+ produce(context, onCompletion = consumes()) {
+ if (n == 0) return@produce
+ require(n >= 0) { "Requested element count $n is less than zero." }
+ var remaining: Int = n
+ for (e in this@take) {
+ send(e)
+ remaining--
+ if (remaining == 0)
+ return@produce
}
}
@@ -704,14 +748,14 @@
* Returns a channel containing first elements satisfying the given [predicate].
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E> ReceiveChannel<E>.takeWhile(context: CoroutineContext = Unconfined, predicate: suspend (E) -> Boolean): ReceiveChannel<E> =
- produce(context) {
- consumeEach {
- if (!predicate(it)) return@produce
- send(it)
+ produce(context, onCompletion = consumes()) {
+ for (e in this@takeWhile) {
+ if (!predicate(e)) return@produce
+ send(e)
}
}
@@ -724,9 +768,9 @@
* The returned map preserves the entry iteration order of the original channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, K, V> ReceiveChannel<E>.associate(transform: (E) -> Pair<K, V>): Map<K, V> =
+public suspend inline fun <E, K, V> ReceiveChannel<E>.associate(transform: (E) -> Pair<K, V>): Map<K, V> =
associateTo(LinkedHashMap(), transform)
/**
@@ -738,9 +782,9 @@
* The returned map preserves the entry iteration order of the original channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, K> ReceiveChannel<E>.associateBy(keySelector: (E) -> K): Map<K, E> =
+public suspend inline fun <E, K> ReceiveChannel<E>.associateBy(keySelector: (E) -> K): Map<K, E> =
associateByTo(LinkedHashMap(), keySelector)
/**
@@ -751,9 +795,9 @@
* The returned map preserves the entry iteration order of the original channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, K, V> ReceiveChannel<E>.associateBy(keySelector: (E) -> K, valueTransform: (E) -> V): Map<K, V> =
+public suspend inline fun <E, K, V> ReceiveChannel<E>.associateBy(keySelector: (E) -> K, valueTransform: (E) -> V): Map<K, V> =
associateByTo(LinkedHashMap(), keySelector, valueTransform)
/**
@@ -764,9 +808,9 @@
* If any two elements would have the same key returned by [keySelector] the last one gets added to the map.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, K, M : MutableMap<in K, in E>> ReceiveChannel<E>.associateByTo(destination: M, keySelector: (E) -> K): M {
+public suspend inline fun <E, K, M : MutableMap<in K, in E>> ReceiveChannel<E>.associateByTo(destination: M, keySelector: (E) -> K): M {
consumeEach {
destination.put(keySelector(it), it)
}
@@ -781,9 +825,9 @@
* If any two elements would have the same key returned by [keySelector] the last one gets added to the map.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, K, V, M : MutableMap<in K, in V>> ReceiveChannel<E>.associateByTo(destination: M, keySelector: (E) -> K, valueTransform: (E) -> V): M {
+public suspend inline fun <E, K, V, M : MutableMap<in K, in V>> ReceiveChannel<E>.associateByTo(destination: M, keySelector: (E) -> K, valueTransform: (E) -> V): M {
consumeEach {
destination.put(keySelector(it), valueTransform(it))
}
@@ -797,9 +841,9 @@
* If any of two pairs would have the same key the last one gets added to the map.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, K, V, M : MutableMap<in K, in V>> ReceiveChannel<E>.associateTo(destination: M, transform: (E) -> Pair<K, V>): M {
+public suspend inline fun <E, K, V, M : MutableMap<in K, in V>> ReceiveChannel<E>.associateTo(destination: M, transform: (E) -> Pair<K, V>): M {
consumeEach {
destination += transform(it)
}
@@ -811,7 +855,7 @@
* and appends the results to the given [destination].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E, C : SendChannel<E>> ReceiveChannel<E>.toChannel(destination: C): C {
consumeEach {
@@ -824,7 +868,7 @@
* Appends all elements to the given [destination] collection.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E, C : MutableCollection<in E>> ReceiveChannel<E>.toCollection(destination: C): C {
consumeEach {
@@ -837,7 +881,7 @@
* Returns a [List] containing all elements.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.toList(): List<E> =
this.toMutableList()
@@ -846,7 +890,7 @@
* Returns a [Map] filled with all elements of this channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <K, V> ReceiveChannel<Pair<K, V>>.toMap(): Map<K, V> =
toMap(LinkedHashMap())
@@ -855,7 +899,7 @@
* Returns a [MutableMap] filled with all elements of this channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <K, V, M : MutableMap<in K, in V>> ReceiveChannel<Pair<K, V>>.toMap(destination: M): M {
consumeEach {
@@ -868,7 +912,7 @@
* Returns a [MutableList] filled with all elements of this channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.toMutableList(): MutableList<E> =
toCollection(ArrayList())
@@ -879,22 +923,22 @@
* The returned set preserves the element iteration order of the original channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.toSet(): Set<E> =
- toCollection(LinkedHashSet())
+ this.toMutableSet()
/**
* Returns a single channel of all elements from results of [transform] function being invoked on each element of original channel.
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E, R> ReceiveChannel<E>.flatMap(context: CoroutineContext = Unconfined, transform: suspend (E) -> ReceiveChannel<R>): ReceiveChannel<R> =
- produce(context) {
- consumeEach {
- transform(it).toChannel(this)
+ produce(context, onCompletion = consumes()) {
+ for (e in this@flatMap) {
+ transform(e).toChannel(this)
}
}
@@ -905,9 +949,9 @@
* The returned map preserves the entry iteration order of the keys produced from the original channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, K> ReceiveChannel<E>.groupBy(keySelector: (E) -> K): Map<K, List<E>> =
+public suspend inline fun <E, K> ReceiveChannel<E>.groupBy(keySelector: (E) -> K): Map<K, List<E>> =
groupByTo(LinkedHashMap(), keySelector)
/**
@@ -918,9 +962,9 @@
* The returned map preserves the entry iteration order of the keys produced from the original channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, K, V> ReceiveChannel<E>.groupBy(keySelector: (E) -> K, valueTransform: (E) -> V): Map<K, List<V>> =
+public suspend inline fun <E, K, V> ReceiveChannel<E>.groupBy(keySelector: (E) -> K, valueTransform: (E) -> V): Map<K, List<V>> =
groupByTo(LinkedHashMap(), keySelector, valueTransform)
/**
@@ -930,9 +974,9 @@
* @return The [destination] map.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, K, M : MutableMap<in K, MutableList<E>>> ReceiveChannel<E>.groupByTo(destination: M, keySelector: (E) -> K): M {
+public suspend inline fun <E, K, M : MutableMap<in K, MutableList<E>>> ReceiveChannel<E>.groupByTo(destination: M, keySelector: (E) -> K): M {
consumeEach {
val key = keySelector(it)
val list = destination.getOrPut(key) { ArrayList() }
@@ -949,9 +993,9 @@
* @return The [destination] map.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, K, V, M : MutableMap<in K, MutableList<V>>> ReceiveChannel<E>.groupByTo(destination: M, keySelector: (E) -> K, valueTransform: (E) -> V): M {
+public suspend inline fun <E, K, V, M : MutableMap<in K, MutableList<V>>> ReceiveChannel<E>.groupByTo(destination: M, keySelector: (E) -> K, valueTransform: (E) -> V): M {
consumeEach {
val key = keySelector(it)
val list = destination.getOrPut(key) { ArrayList() }
@@ -965,11 +1009,11 @@
* to each element in the original channel.
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
+// todo: mark transform with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E, R> ReceiveChannel<E>.map(context: CoroutineContext = Unconfined, transform: suspend (E) -> R): ReceiveChannel<R> =
- produce(context) {
+ produce(context, onCompletion = consumes()) {
consumeEach {
send(transform(it))
}
@@ -982,14 +1026,14 @@
* and returns the result of the transform applied to the element.
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E, R> ReceiveChannel<E>.mapIndexed(context: CoroutineContext = Unconfined, transform: suspend (index: Int, E) -> R): ReceiveChannel<R> =
- produce(context) {
+ produce(context, onCompletion = consumes()) {
var index = 0
- consumeEach {
- send(transform(index++, it))
+ for (e in this@mapIndexed) {
+ send(transform(index++, e))
}
}
@@ -1000,7 +1044,7 @@
* and returns the result of the transform applied to the element.
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E, R : Any> ReceiveChannel<E>.mapIndexedNotNull(context: CoroutineContext = Unconfined, transform: suspend (index: Int, E) -> R?): ReceiveChannel<R> =
@@ -1013,9 +1057,9 @@
* and returns the result of the transform applied to the element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, R : Any, C : MutableCollection<in R>> ReceiveChannel<E>.mapIndexedNotNullTo(destination: C, transform: (index: Int, E) -> R?): C {
+public suspend inline fun <E, R : Any, C : MutableCollection<in R>> ReceiveChannel<E>.mapIndexedNotNullTo(destination: C, transform: (index: Int, E) -> R?): C {
consumeEachIndexed { (index, element) ->
transform(index, element)?.let { destination.add(it) }
}
@@ -1029,9 +1073,9 @@
* and returns the result of the transform applied to the element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, R : Any, C : SendChannel<R>> ReceiveChannel<E>.mapIndexedNotNullTo(destination: C, transform: (index: Int, E) -> R?): C {
+public suspend inline fun <E, R : Any, C : SendChannel<R>> ReceiveChannel<E>.mapIndexedNotNullTo(destination: C, transform: (index: Int, E) -> R?): C {
consumeEachIndexed { (index, element) ->
transform(index, element)?.let { destination.send(it) }
}
@@ -1045,9 +1089,9 @@
* and returns the result of the transform applied to the element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, R, C : MutableCollection<in R>> ReceiveChannel<E>.mapIndexedTo(destination: C, transform: (index: Int, E) -> R): C {
+public suspend inline fun <E, R, C : MutableCollection<in R>> ReceiveChannel<E>.mapIndexedTo(destination: C, transform: (index: Int, E) -> R): C {
var index = 0
consumeEach {
destination.add(transform(index++, it))
@@ -1062,9 +1106,9 @@
* and returns the result of the transform applied to the element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, R, C : SendChannel<R>> ReceiveChannel<E>.mapIndexedTo(destination: C, transform: (index: Int, E) -> R): C {
+public suspend inline fun <E, R, C : SendChannel<R>> ReceiveChannel<E>.mapIndexedTo(destination: C, transform: (index: Int, E) -> R): C {
var index = 0
consumeEach {
destination.send(transform(index++, it))
@@ -1077,7 +1121,7 @@
* to each element in the original channel.
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E, R : Any> ReceiveChannel<E>.mapNotNull(context: CoroutineContext = Unconfined, transform: suspend (E) -> R?): ReceiveChannel<R> =
@@ -1088,9 +1132,9 @@
* and appends only the non-null results to the given [destination].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, R : Any, C : MutableCollection<in R>> ReceiveChannel<E>.mapNotNullTo(destination: C, transform: (E) -> R?): C {
+public suspend inline fun <E, R : Any, C : MutableCollection<in R>> ReceiveChannel<E>.mapNotNullTo(destination: C, transform: (E) -> R?): C {
consumeEach {
transform(it)?.let { destination.add(it) }
}
@@ -1102,9 +1146,9 @@
* and appends only the non-null results to the given [destination].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, R : Any, C : SendChannel<R>> ReceiveChannel<E>.mapNotNullTo(destination: C, transform: (E) -> R?): C {
+public suspend inline fun <E, R : Any, C : SendChannel<R>> ReceiveChannel<E>.mapNotNullTo(destination: C, transform: (E) -> R?): C {
consumeEach {
transform(it)?.let { destination.send(it) }
}
@@ -1116,9 +1160,9 @@
* and appends the results to the given [destination].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, R, C : MutableCollection<in R>> ReceiveChannel<E>.mapTo(destination: C, transform: (E) -> R): C {
+public suspend inline fun <E, R, C : MutableCollection<in R>> ReceiveChannel<E>.mapTo(destination: C, transform: (E) -> R): C {
consumeEach {
destination.add(transform(it))
}
@@ -1130,9 +1174,9 @@
* and appends the results to the given [destination].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, R, C : SendChannel<R>> ReceiveChannel<E>.mapTo(destination: C, transform: (E) -> R): C {
+public suspend inline fun <E, R, C : SendChannel<R>> ReceiveChannel<E>.mapTo(destination: C, transform: (E) -> R): C {
consumeEach {
destination.send(transform(it))
}
@@ -1143,13 +1187,13 @@
* Returns a channel of [IndexedValue] for each element of the original channel.
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public fun <E> ReceiveChannel<E>.withIndex(context: CoroutineContext = Unconfined): ReceiveChannel<IndexedValue<E>> =
- produce(context) {
+ produce(context, onCompletion = consumes()) {
var index = 0
- consumeEach {
- send(IndexedValue(index++, it))
+ for (e in this@withIndex) {
+ send(IndexedValue(index++, e))
}
}
@@ -1159,7 +1203,7 @@
* The elements in the resulting channel are in the same order as they were in the source channel.
*
* The operation is _intermediate_ and _stateful_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public fun <E> ReceiveChannel<E>.distinct(): ReceiveChannel<E> =
this.distinctBy { it }
@@ -1171,16 +1215,16 @@
* The elements in the resulting channel are in the same order as they were in the source channel.
*
* The operation is _intermediate_ and _stateful_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E, K> ReceiveChannel<E>.distinctBy(context: CoroutineContext = Unconfined, selector: suspend (E) -> K): ReceiveChannel<E> =
- produce(context) {
+ produce(context, onCompletion = consumes()) {
val keys = HashSet<K>()
- consumeEach {
- val k = selector(it)
+ for (e in this@distinctBy) {
+ val k = selector(e)
if (k !in keys) {
- send(it)
+ send(e)
keys += k
}
}
@@ -1192,23 +1236,18 @@
* The returned set preserves the element iteration order of the original channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public suspend fun <E> ReceiveChannel<E>.toMutableSet(): MutableSet<E> {
- val set = LinkedHashSet<E>()
- consumeEach {
- set.add(it)
- }
- return set
-}
+public suspend fun <E> ReceiveChannel<E>.toMutableSet(): MutableSet<E> =
+ toCollection(LinkedHashSet())
/**
* Returns `true` if all elements match the given [predicate].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.all(predicate: (E) -> Boolean): Boolean {
+public suspend inline fun <E> ReceiveChannel<E>.all(predicate: (E) -> Boolean): Boolean {
consumeEach {
if (!predicate(it)) return false
}
@@ -1219,7 +1258,7 @@
* Returns `true` if channel has at least one element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.any(): Boolean =
consume {
@@ -1230,9 +1269,9 @@
* Returns `true` if at least one element matches the given [predicate].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.any(predicate: (E) -> Boolean): Boolean {
+public suspend inline fun <E> ReceiveChannel<E>.any(predicate: (E) -> Boolean): Boolean {
consumeEach {
if (predicate(it)) return true
}
@@ -1243,7 +1282,7 @@
* Returns the number of elements in this channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.count(): Int {
var count = 0
@@ -1255,9 +1294,9 @@
* Returns the number of elements matching the given [predicate].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.count(predicate: (E) -> Boolean): Int {
+public suspend inline fun <E> ReceiveChannel<E>.count(predicate: (E) -> Boolean): Int {
var count = 0
consumeEach {
if (predicate(it)) count++
@@ -1269,9 +1308,9 @@
* Accumulates value starting with [initial] value and applying [operation] from left to right to current accumulator value and each element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, R> ReceiveChannel<E>.fold(initial: R, operation: (acc: R, E) -> R): R {
+public suspend inline fun <E, R> ReceiveChannel<E>.fold(initial: R, operation: (acc: R, E) -> R): R {
var accumulator = initial
consumeEach {
accumulator = operation(accumulator, it)
@@ -1286,9 +1325,9 @@
* and the element itself, and calculates the next accumulator value.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, R> ReceiveChannel<E>.foldIndexed(initial: R, operation: (index: Int, acc: R, E) -> R): R {
+public suspend inline fun <E, R> ReceiveChannel<E>.foldIndexed(initial: R, operation: (index: Int, acc: R, E) -> R): R {
var index = 0
var accumulator = initial
consumeEach {
@@ -1301,9 +1340,9 @@
* Returns the first element yielding the largest value of the given function or `null` if there are no elements.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, R : Comparable<R>> ReceiveChannel<E>.maxBy(selector: (E) -> R): E? =
+public suspend inline fun <E, R : Comparable<R>> ReceiveChannel<E>.maxBy(selector: (E) -> R): E? =
consume {
val iterator = iterator()
if (!iterator.hasNext()) return null
@@ -1324,7 +1363,7 @@
* Returns the first element having the largest value according to the provided [comparator] or `null` if there are no elements.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.maxWith(comparator: Comparator<in E>): E? =
consume {
@@ -1342,9 +1381,9 @@
* Returns the first element yielding the smallest value of the given function or `null` if there are no elements.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E, R : Comparable<R>> ReceiveChannel<E>.minBy(selector: (E) -> R): E? =
+public suspend inline fun <E, R : Comparable<R>> ReceiveChannel<E>.minBy(selector: (E) -> R): E? =
consume {
val iterator = iterator()
if (!iterator.hasNext()) return null
@@ -1365,7 +1404,7 @@
* Returns the first element having the smallest value according to the provided [comparator] or `null` if there are no elements.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.minWith(comparator: Comparator<in E>): E? =
consume {
@@ -1383,7 +1422,7 @@
* Returns `true` if the channel has no elements.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public suspend fun <E> ReceiveChannel<E>.none(): Boolean =
consume {
@@ -1394,9 +1433,9 @@
* Returns `true` if no elements match the given [predicate].
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.none(predicate: (E) -> Boolean): Boolean {
+public suspend inline fun <E> ReceiveChannel<E>.none(predicate: (E) -> Boolean): Boolean {
consumeEach {
if (predicate(it)) return false
}
@@ -1407,9 +1446,9 @@
* Accumulates value starting with the first element and applying [operation] from left to right to current accumulator value and each element.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <S, E : S> ReceiveChannel<E>.reduce(operation: (acc: S, E) -> S): S =
+public suspend inline fun <S, E : S> ReceiveChannel<E>.reduce(operation: (acc: S, E) -> S): S =
consume {
val iterator = this.iterator()
if (!iterator.hasNext()) throw UnsupportedOperationException("Empty channel can't be reduced.")
@@ -1427,9 +1466,10 @@
* and the element itself and calculates the next accumulator value.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <S, E : S> ReceiveChannel<E>.reduceIndexed(operation: (index: Int, acc: S, E) -> S): S =
+// todo: mark operation with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
+public suspend inline fun <S, E : S> ReceiveChannel<E>.reduceIndexed(operation: (index: Int, acc: S, E) -> S): S =
consume {
val iterator = this.iterator()
if (!iterator.hasNext()) throw UnsupportedOperationException("Empty channel can't be reduced.")
@@ -1445,9 +1485,9 @@
* Returns the sum of all values produced by [selector] function applied to each element in the channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.sumBy(selector: (E) -> Int): Int {
+public suspend inline fun <E> ReceiveChannel<E>.sumBy(selector: (E) -> Int): Int {
var sum = 0
consumeEach {
sum += selector(it)
@@ -1459,9 +1499,9 @@
* Returns the sum of all values produced by [selector] function applied to each element in the channel.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.sumByDouble(selector: (E) -> Double): Double {
+public suspend inline fun <E> ReceiveChannel<E>.sumByDouble(selector: (E) -> Double): Double {
var sum = 0.0
consumeEach {
sum += selector(it)
@@ -1473,7 +1513,7 @@
* Returns an original collection containing all the non-`null` elements, throwing an [IllegalArgumentException] if there are any `null` elements.
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
public fun <E : Any> ReceiveChannel<E?>.requireNoNulls(): ReceiveChannel<E> =
map { it ?: throw IllegalArgumentException("null element found in $this.") }
@@ -1484,9 +1524,9 @@
* while *second* list contains elements for which [predicate] yielded `false`.
*
* The operation is _terminal_.
- * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
*/
-public inline suspend fun <E> ReceiveChannel<E>.partition(predicate: (E) -> Boolean): Pair<List<E>, List<E>> {
+public suspend inline fun <E> ReceiveChannel<E>.partition(predicate: (E) -> Boolean): Pair<List<E>, List<E>> {
val first = ArrayList<E>()
val second = ArrayList<E>()
consumeEach {
@@ -1504,7 +1544,7 @@
* Resulting channel has length of shortest input channel.
*
* The operation is _intermediate_ and _stateless_.
- * This function [consumes][consume] all elements of both the original [ReceiveChannel] and the `other` one.
+ * This function [consumes][ReceiveChannel.consume] all elements of both the original [ReceiveChannel] and the `other` one.
*/
public infix fun <E, R> ReceiveChannel<E>.zip(other: ReceiveChannel<R>): ReceiveChannel<Pair<E, R>> =
zip(other) { t1, t2 -> t1 to t2 }
@@ -1515,15 +1555,15 @@
* The operation is _intermediate_ and _stateless_.
* This function [consumes][consume] all elements of both the original [ReceiveChannel] and the `other` one.
*/
+// todo: mark transform with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E, R, V> ReceiveChannel<E>.zip(other: ReceiveChannel<R>, context: CoroutineContext = Unconfined, transform: (a: E, b: R) -> V): ReceiveChannel<V> =
- produce(context) {
- other.consume {
- val otherIterator = other.iterator()
- this@zip.consumeEach { element1 ->
- if (!otherIterator.hasNext()) return@consumeEach
- val element2 = otherIterator.next()
- send(transform(element1, element2))
- }
+ produce(context, onCompletion = consumesAll(this, other)) {
+ val otherIterator = other.iterator()
+ this@zip.consumeEach { element1 ->
+ if (!otherIterator.hasNext()) return@consumeEach
+ val element2 = otherIterator.next()
+ send(transform(element1, element2))
}
}
+
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Produce.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Produce.kt
index 61e6a6f..1565ecd 100644
--- a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Produce.kt
+++ b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Produce.kt
@@ -70,17 +70,20 @@
* @param context context of the coroutine. The default value is [DefaultDispatcher].
* @param capacity capacity of the channel's buffer (no buffer by default).
* @param parent explicitly specifies the parent job, overrides job from the [context] (if any).*
+ * @param onCompletion optional completion handler for the producer coroutine (see [Job.invokeOnCompletion]).
* @param block the coroutine code.
*/
public fun <E> produce(
context: CoroutineContext = DefaultDispatcher,
capacity: Int = 0,
parent: Job? = null,
+ onCompletion: CompletionHandler? = null,
block: suspend ProducerScope<E>.() -> Unit
): ReceiveChannel<E> {
val channel = Channel<E>(capacity)
val newContext = newCoroutineContext(context, parent)
val coroutine = ProducerCoroutine(newContext, channel)
+ if (onCompletion != null) coroutine.invokeOnCompletion(handler = onCompletion)
coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
return coroutine
}
@@ -90,6 +93,15 @@
public fun <E> produce(
context: CoroutineContext = DefaultDispatcher,
capacity: Int = 0,
+ parent: Job? = null,
+ block: suspend ProducerScope<E>.() -> Unit
+): ReceiveChannel<E> = produce(context, capacity, parent, block = block)
+
+/** @suppress **Deprecated**: Binary compatibility */
+@Deprecated(message = "Binary compatibility", level = DeprecationLevel.HIDDEN)
+public fun <E> produce(
+ context: CoroutineContext = DefaultDispatcher,
+ capacity: Int = 0,
block: suspend ProducerScope<E>.() -> Unit
): ProducerJob<E> =
produce(context, capacity, block = block) as ProducerJob<E>