Introduced consumeEach for channels and reactive streams, deprecated iteration on reactive streams

Author: elizarov

coroutines-guide.md

@@ -1190,7 +1190,8 @@ This is a part of _producer-consumer_ pattern that is often found in concurrent
 You could abstract such a producer into a function that takes channel as its parameter, but this goes contrary
 to common sense that results must be returned from functions. 
 
-There is a convenience coroutine builder named [produce] that makes it easy to do it right:
+There is a convenience coroutine builder named [produce] that makes it easy to do it right on producer side,
+and an extension function [consumeEach], that can replace a `for` loop on the consumer side:
 
 ```kotlin
 fun produceSquares() = produce<Int>(CommonPool) {
@@ -1199,7 +1200,7 @@ fun produceSquares() = produce<Int>(CommonPool) {
 
 fun main(args: Array<String>) = runBlocking<Unit> {
     val squares = produceSquares()
-    for (y in squares) println(y)
+    squares.consumeEach { println(it) }
     println("Done!")
 }
 ```
@@ -1363,8 +1364,8 @@ received number:
 
 ```kotlin
 fun launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch(CommonPool) {
-    for (x in channel) {
-        println("Processor #$id received $x")
+    channel.consumeEach {
+        println("Processor #$id received $it")
     }    
 }
 ```
@@ -1988,10 +1989,10 @@ Consumer is going to be quite slow, taking 250 ms to process each number:
 fun main(args: Array<String>) = runBlocking<Unit> {
     val side = Channel<Int>() // allocate side channel
     launch(context) { // this is a very fast consumer for the side channel
-        for (num in side) println("Side channel has $num")
+        side.consumeEach { println("Side channel has $it") }
     }
-    for (num in produceNumbers(side)) {
-        println("Consuming $num")
+    produceNumbers(side).consumeEach { 
+        println("Consuming $it")
         delay(250) // let us digest the consumed number properly, do not hurry
     }
     println("Done consuming")
@@ -2193,6 +2194,7 @@ Channel was closed
 [ReceiveChannel.receive]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-receive-channel/receive.html
 [SendChannel.close]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-send-channel/close.html
 [produce]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/produce.html
+[consumeEach]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/consume-each.html
 [Channel.invoke]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-channel/invoke.html
 [actor]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/actor.html
 <!--- INDEX kotlinx.coroutines.experimental.selects -->

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

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental.channels
+
+/**
+ * Performs the given [action] on each received element.
+ */
+// :todo: make it inline when this bug is fixed: https://youtrack.jetbrains.com/issue/KT-16448
+public suspend fun <E> ReceiveChannel<E>.consumeEach(action: suspend (E) -> Unit) {
+    for (element in this) action(element)
+}

kotlinx-coroutines-core/src/test/kotlin/guide/example-channel-03.kt

@@ -17,15 +17,17 @@
 // This file was automatically generated from coroutines-guide.md by Knit tool. Do not edit.
 package guide.channel.example03
 
-import kotlinx.coroutines.experimental.*
-import kotlinx.coroutines.experimental.channels.*
+import kotlinx.coroutines.experimental.CommonPool
+import kotlinx.coroutines.experimental.channels.consumeEach
+import kotlinx.coroutines.experimental.channels.produce
+import kotlinx.coroutines.experimental.runBlocking
 
 fun produceSquares() = produce<Int>(CommonPool) {
     for (x in 1..5) send(x * x)
 }
 
 fun main(args: Array<String>) = runBlocking<Unit> {
     val squares = produceSquares()
-    for (y in squares) println(y)
+    squares.consumeEach { println(it) }
     println("Done!")
 }

kotlinx-coroutines-core/src/test/kotlin/guide/example-channel-06.kt

@@ -17,8 +17,13 @@
 // This file was automatically generated from coroutines-guide.md by Knit tool. Do not edit.
 package guide.channel.example06
 
-import kotlinx.coroutines.experimental.*
-import kotlinx.coroutines.experimental.channels.*
+import kotlinx.coroutines.experimental.CommonPool
+import kotlinx.coroutines.experimental.channels.ReceiveChannel
+import kotlinx.coroutines.experimental.channels.consumeEach
+import kotlinx.coroutines.experimental.channels.produce
+import kotlinx.coroutines.experimental.delay
+import kotlinx.coroutines.experimental.launch
+import kotlinx.coroutines.experimental.runBlocking
 
 fun produceNumbers() = produce<Int>(CommonPool) {
     var x = 1 // start from 1
@@ -29,8 +34,8 @@ fun produceNumbers() = produce<Int>(CommonPool) {
 }
 
 fun launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch(CommonPool) {
-    for (x in channel) {
-        println("Processor #$id received $x")
+    channel.consumeEach {
+        println("Processor #$id received $it")
     }    
 }
 

kotlinx-coroutines-core/src/test/kotlin/guide/example-select-03.kt

@@ -17,9 +17,15 @@
 // This file was automatically generated from coroutines-guide.md by Knit tool. Do not edit.
 package guide.select.example03
 
-import kotlinx.coroutines.experimental.*
-import kotlinx.coroutines.experimental.channels.*
-import kotlinx.coroutines.experimental.selects.*
+import kotlinx.coroutines.experimental.CommonPool
+import kotlinx.coroutines.experimental.channels.Channel
+import kotlinx.coroutines.experimental.channels.SendChannel
+import kotlinx.coroutines.experimental.channels.consumeEach
+import kotlinx.coroutines.experimental.channels.produce
+import kotlinx.coroutines.experimental.delay
+import kotlinx.coroutines.experimental.launch
+import kotlinx.coroutines.experimental.runBlocking
+import kotlinx.coroutines.experimental.selects.select
 
 fun produceNumbers(side: SendChannel<Int>) = produce<Int>(CommonPool) {
     for (num in 1..10) { // produce 10 numbers from 1 to 10
@@ -34,10 +40,10 @@ fun produceNumbers(side: SendChannel<Int>) = produce<Int>(CommonPool) {
 fun main(args: Array<String>) = runBlocking<Unit> {
     val side = Channel<Int>() // allocate side channel
     launch(context) { // this is a very fast consumer for the side channel
-        for (num in side) println("Side channel has $num")
+        side.consumeEach { println("Side channel has $it") }
     }
-    for (num in produceNumbers(side)) {
-        println("Consuming $num")
+    produceNumbers(side).consumeEach { 
+        println("Consuming $it")
         delay(250) // let us digest the consumed number properly, do not hurry
     }
     println("Done consuming")