Introduce deprecated ReceiveChannel#use to preserve backward compatibility with SubscriptionReceiveChannel, fix reactive samples

Author: qwwdfsad

binary-compatibility-validator/build.gradle

@@ -43,6 +43,7 @@ sourceSets {
 }
 
 test {
+    dependsOn cleanCompileTestKotlin
     dependsOn configurations.testArtifacts
 
     systemProperty 'testCasesClassesDirs', sourceSets.test.output.classesDirs.asPath

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt

@@ -581,6 +581,7 @@ public final class kotlinx/coroutines/experimental/channels/BroadcastChannel$Fac
 
 public final class kotlinx/coroutines/experimental/channels/BroadcastChannelKt {
 	public static final fun BroadcastChannel (I)Lkotlinx/coroutines/experimental/channels/BroadcastChannel;
+	public static final synthetic fun use (Lkotlinx/coroutines/experimental/channels/ReceiveChannel;Lkotlin/jvm/functions/Function1;)V
 }
 
 public abstract interface class kotlinx/coroutines/experimental/channels/Channel : kotlinx/coroutines/experimental/channels/ReceiveChannel, kotlinx/coroutines/experimental/channels/SendChannel {

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

@@ -97,3 +97,18 @@ public interface SubscriptionReceiveChannel<out T> : ReceiveChannel<T>, Closeabl
     @Deprecated("Use `cancel`", replaceWith = ReplaceWith("cancel()"))
     public override fun close() { cancel() }
 }
+
+/** @suppress **Deprecated**: Left here for migration from SubscriptionReceiveChannel */
+@Deprecated(level = DeprecationLevel.HIDDEN, message = "Use `ReceiveChannel<*>.consume` instead")
+public inline fun <E, R> ReceiveChannel<E>.use(block: (ReceiveChannel<E>) -> R) {
+    var exception: Throwable? = null
+    try {
+        block(this)
+    } catch (t: Throwable) {
+        exception = t
+        throw t
+    }
+    finally {
+        this.cancel(exception)
+    }
+}

reactive/coroutines-guide-reactive.md

@@ -237,8 +237,8 @@ as shown in the following example:
 <!--- INCLUDE
 import io.reactivex.*
 import kotlinx.coroutines.experimental.*
+import kotlinx.coroutines.experimental.channels.*
 import kotlinx.coroutines.experimental.reactive.*
-import kotlin.coroutines.experimental.*
 -->
 
 ```kotlin
@@ -247,8 +247,8 @@ fun main(args: Array<String>) = runBlocking<Unit> {
         .doOnSubscribe { println("OnSubscribe") } // provide some insight
         .doFinally { println("Finally") }         // ... into what's going on
     var cnt = 0 
-    source.openSubscription().use { channel -> // open channel to the source
-        for (x in channel) { // iterate over the channel to receive elements from it
+    source.openSubscription().consume { // open channel to the source
+        for (x in this) { // iterate over the channel to receive elements from it
             println(x)
             if (++cnt >= 3) break // break when 3 elements are printed
         }
@@ -677,6 +677,7 @@ We need to relay all the elements from the source stream until the other stream
 emits anything. However, we have [select] expression to rescue us in coroutines implementation:
 
 <!--- INCLUDE
+import kotlinx.coroutines.experimental.channels.*
 import kotlinx.coroutines.experimental.*
 import kotlinx.coroutines.experimental.reactive.*
 import kotlinx.coroutines.experimental.selects.*
@@ -686,11 +687,13 @@ import kotlin.coroutines.experimental.*
 
 ```kotlin
 fun <T, U> Publisher<T>.takeUntil(context: CoroutineContext, other: Publisher<U>) = publish<T>(context) {
-    this@takeUntil.openSubscription().use { thisChannel -> // explicitly open channel to Publisher<T>
-        other.openSubscription().use { otherChannel ->     // explicitly open channel to Publisher<U>
+    this@takeUntil.openSubscription().consume { // explicitly open channel to Publisher<T>
+        val current = this
+        other.openSubscription().consume { // explicitly open channel to Publisher<U>
+            val other = this
             whileSelect {
-                otherChannel.onReceive { false }          // bail out on any received element from `other`
-                thisChannel.onReceive { send(it); true }  // resend element from this channel and continue
+                other.onReceive { false }          // bail out on any received element from `other`
+                current.onReceive { send(it); true }  // resend element from this channel and continue
             }
         }
     }

reactive/kotlinx-coroutines-rx2/src/test/kotlin/guide/example-reactive-basic-03.kt

@@ -19,18 +19,19 @@ package guide.reactive.basic.example03
 
 import io.reactivex.*
 import kotlinx.coroutines.experimental.*
+import kotlinx.coroutines.experimental.channels.*
 import kotlinx.coroutines.experimental.reactive.*
-import kotlin.coroutines.experimental.*
 
 fun main(args: Array<String>) = runBlocking<Unit> {
     val source = Flowable.range(1, 5) // a range of five numbers
         .doOnSubscribe { println("OnSubscribe") } // provide some insight
         .doFinally { println("Finally") }         // ... into what's going on
     var cnt = 0 
-    val channel = source.openSubscription() // open channel to the source
-    for (x in channel) { // iterate over the channel to receive elements from it
-        println(x)
-        if (++cnt >= 3) break // break when 3 elements are printed
+    source.openSubscription().consume { // open channel to the source
+        for (x in this) { // iterate over the channel to receive elements from it
+            println(x)
+            if (++cnt >= 3) break // break when 3 elements are printed
+        }
+        // `use` will close the channel when this block of code is complete
     }
-    channel.cancel()    // `cancel` closes the channel
 }

reactive/kotlinx-coroutines-rx2/src/test/kotlin/guide/example-reactive-operators-03.kt

@@ -17,21 +17,24 @@
 // This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
 package guide.reactive.operators.example03
 
+import kotlinx.coroutines.experimental.channels.*
 import kotlinx.coroutines.experimental.*
 import kotlinx.coroutines.experimental.reactive.*
 import kotlinx.coroutines.experimental.selects.*
 import org.reactivestreams.*
 import kotlin.coroutines.experimental.*
 
 fun <T, U> Publisher<T>.takeUntil(context: CoroutineContext, other: Publisher<U>) = publish<T>(context) {
-    val thisChannel = this@takeUntil.openSubscription() // explicitly open channel to Publisher<T>
-    val otherChannel = other.openSubscription() // explicitly open channel to Publisher<U>
-    whileSelect {
-        otherChannel.onReceive { false }          // bail out on any received element from `other`
-        thisChannel.onReceive { send(it); true }  // resend element from this channel and continue
+    this@takeUntil.openSubscription().consume { // explicitly open channel to Publisher<T>
+        val current = this
+        other.openSubscription().consume { // explicitly open channel to Publisher<U>
+            val other = this
+            whileSelect {
+                other.onReceive { false }          // bail out on any received element from `other`
+                current.onReceive { send(it); true }  // resend element from this channel and continue
+            }
+        }
     }
-    thisChannel.cancel()
-    otherChannel.cancel()
 }
 
 fun rangeWithInterval(context: CoroutineContext, time: Long, start: Int, count: Int) = publish<Int>(context) {