Replace SubscriptionReceiveChannel with ReceiveChannel

Fixes #283

Author: deva666

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

@@ -68,7 +68,7 @@ class ArrayBroadcastChannel<E>(
     override val isBufferAlwaysFull: Boolean get() = false
     override val isBufferFull: Boolean get() = size >= capacity
 
-    override fun openSubscription(): SubscriptionReceiveChannel<E> =
+    override fun openSubscription(): ReceiveChannel<E> =
         Subscriber(this).also {
             updateHead(addSub = it)
         }
@@ -195,7 +195,7 @@ class ArrayBroadcastChannel<E>(
 
     private class Subscriber<E>(
         private val broadcastChannel: ArrayBroadcastChannel<E>
-    ) : AbstractChannel<E>(), SubscriptionReceiveChannel<E> {
+    ) : AbstractChannel<E>(), ReceiveChannel<E> {
         private val subLock = ReentrantLock()
 
         @Volatile

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

@@ -23,7 +23,7 @@ import kotlinx.coroutines.experimental.internal.Closeable
 
 /**
  * Broadcast channel is a non-blocking primitive for communication between the sender and multiple receivers
- * that subscribe for the elements using [openSubscription] function and unsubscribe using [SubscriptionReceiveChannel.close]
+ * that subscribe for the elements using [openSubscription] function and unsubscribe using [ReceiveChannel.cancel]
  * function.
  *
  * See `BroadcastChannel()` factory function for the description of available
@@ -45,17 +45,17 @@ public interface BroadcastChannel<E> : SendChannel<E> {
 
     /**
      * Subscribes to this [BroadcastChannel] and returns a channel to receive elements from it.
-     * The resulting channel shall be [closed][SubscriptionReceiveChannel.close] to unsubscribe from this
+     * The resulting channel shall be [cancelled][ReceiveChannel.cancel] to unsubscribe from this
      * broadcast channel.
      */
-    public fun openSubscription(): SubscriptionReceiveChannel<E>
+    public fun openSubscription(): ReceiveChannel<E>
 
     /**
      * @suppress **Deprecated**: Renamed to [openSubscription]
      */
     @Deprecated(message = "Renamed to `openSubscription`",
         replaceWith = ReplaceWith("openSubscription()"))
-    public fun open(): SubscriptionReceiveChannel<E> = openSubscription()
+    public fun open(): ReceiveChannel<E> = openSubscription()
 }
 
 /**
@@ -80,6 +80,7 @@ public fun <E> BroadcastChannel(capacity: Int): BroadcastChannel<E> =
  *
  * Note, that invocation of [cancel] also closes subscription.
  */
+@Deprecated("Deprecated in favour of `ReceiveChannel`")
 public interface SubscriptionReceiveChannel<out T> : ReceiveChannel<T>, Closeable {
     /**
      * Closes this subscription. This is a synonym for [cancel].

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

@@ -46,9 +46,9 @@ public fun <E> Sequence<E>.asReceiveChannel(context: CoroutineContext = Unconfin
 
 /**
  * Opens subscription to this [BroadcastChannel] and makes sure that the given [block] consumes all elements
- * from it by always invoking [cancel][SubscriptionReceiveChannel.cancel] after the execution of the block.
+ * from it by always invoking [cancel][ReceiveChannel.cancel] after the execution of the block.
  */
-public inline fun <E, R> BroadcastChannel<E>.consume(block: SubscriptionReceiveChannel<E>.() -> R): R {
+public inline fun <E, R> BroadcastChannel<E>.consume(block: ReceiveChannel<E>.() -> R): R {
     val channel = openSubscription()
     try {
         return channel.block()

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

@@ -34,7 +34,7 @@ import kotlinx.coroutines.experimental.selects.*
  * This channel is also created by `BroadcastChannel(Channel.CONFLATED)` factory function invocation.
  *
  * This implementation is fully lock-free. In this implementation
- * [opening][openSubscription] and [closing][SubscriptionReceiveChannel.close] subscription takes O(N) time, where N is the
+ * [opening][openSubscription] and [closing][ReceiveChannel.cancel] subscription takes O(N) time, where N is the
  * number of subscribers.
  */
 public class ConflatedBroadcastChannel<E>() : BroadcastChannel<E> {
@@ -114,7 +114,7 @@ public class ConflatedBroadcastChannel<E>() : BroadcastChannel<E> {
     override val isFull: Boolean get() = false
 
     @Suppress("UNCHECKED_CAST")
-    override fun openSubscription(): SubscriptionReceiveChannel<E> {
+    override fun openSubscription(): ReceiveChannel<E> {
         val subscriber = Subscriber<E>(this)
         _state.loop { state ->
             when (state) {
@@ -247,7 +247,7 @@ public class ConflatedBroadcastChannel<E>() : BroadcastChannel<E> {
 
     private class Subscriber<E>(
         private val broadcastChannel: ConflatedBroadcastChannel<E>
-    ) : ConflatedChannel<E>(), SubscriptionReceiveChannel<E> {
+    ) : ConflatedChannel<E>(), ReceiveChannel<E> {
         override fun cancel(cause: Throwable?): Boolean =
             close(cause).also { closed ->
                 if (closed) broadcastChannel.closeSubscriber(this)

common/kotlinx-coroutines-core-common/src/test/kotlin/kotlinx/coroutines/experimental/channels/ConflatedBroadcastChannelTest.kt

@@ -38,7 +38,7 @@ class ConflatedBroadcastChannelTest : TestBase() {
             expect(6)
             assertEquals("two", sub.receive()) // suspends
             expect(12)
-            sub.close()
+            sub.cancel()
             expect(13)
         }
 
@@ -60,7 +60,7 @@ class ConflatedBroadcastChannelTest : TestBase() {
             expect(17)
             assertNull(sub.receiveOrNull()) // suspends until closed
             expect(20)
-            sub.close()
+            sub.cancel()
             expect(21)
         }
 

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/BroadcastChannelMultiReceiveStressTest.kt

@@ -78,15 +78,15 @@ class BroadcastChannelMultiReceiveStressTest(
             val name = "Receiver$receiverIndex"
             println("Launching $name")
             receivers += launch(ctx + CoroutineName(name)) {
-                broadcast.openSubscription().use { sub ->
+                val channel = broadcast.openSubscription()
                     when (receiverIndex % 5) {
-                        0 -> doReceive(sub, receiverIndex)
-                        1 -> doReceiveOrNull(sub, receiverIndex)
-                        2 -> doIterator(sub, receiverIndex)
-                        3 -> doReceiveSelect(sub, receiverIndex)
-                        4 -> doReceiveSelectOrNull(sub, receiverIndex)
+                        0 -> doReceive(channel, receiverIndex)
+                        1 -> doReceiveOrNull(channel, receiverIndex)
+                        2 -> doIterator(channel, receiverIndex)
+                        3 -> doReceiveSelect(channel, receiverIndex)
+                        4 -> doReceiveSelectOrNull(channel, receiverIndex)
                     }
-                }
+                channel.cancel()
             }
             printProgress()
         }

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/BroadcastChannelSubStressTest.kt

@@ -60,13 +60,13 @@ class BroadcastChannelSubStressTest(
             launch(context = ctx + CoroutineName("Receiver")) {
                 var last = -1L
                 while (isActive) {
-                    broadcast.openSubscription().use { sub ->
-                        val i = sub.receive()
-                        check(i >= last) { "Last was $last, got $i" }
-                        if (!kind.isConflated) check(i != last) { "Last was $last, got it again" }
-                        receivedTotal.incrementAndGet()
-                        last = i
-                    }
+                    val channel = broadcast.openSubscription()
+                    val i = channel.receive()
+                    check(i >= last) { "Last was $last, got $i" }
+                    if (!kind.isConflated) check(i != last) { "Last was $last, got it again" }
+                    receivedTotal.incrementAndGet()
+                    last = i
+                    channel.cancel()
                 }
             }
         var prevSent = -1L

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/ConflatedBroadcastChannelNotifyStressTest.kt

@@ -95,7 +95,9 @@ class ConflatedBroadcastChannelNotifyStressTest : TestBase() {
     }
 
     suspend fun waitForEvent(): Int =
-        broadcast.openSubscription().use {
-            it.receive()
+        with(broadcast.openSubscription()) {
+            val value = receive()
+            cancel()
+            value
         }
 }

reactive/kotlinx-coroutines-reactive/src/main/kotlin/kotlinx/coroutines/experimental/reactive/Channel.kt

@@ -19,18 +19,18 @@ package kotlinx.coroutines.experimental.reactive
 import kotlinx.atomicfu.atomic
 import kotlinx.atomicfu.loop
 import kotlinx.coroutines.experimental.channels.LinkedListChannel
-import kotlinx.coroutines.experimental.channels.SubscriptionReceiveChannel
+import kotlinx.coroutines.experimental.channels.ReceiveChannel
 import org.reactivestreams.Publisher
 import org.reactivestreams.Subscriber
 import org.reactivestreams.Subscription
 
 /**
  * Subscribes to this [Publisher] and returns a channel to receive elements emitted by it.
- * The resulting channel shall be [closed][SubscriptionReceiveChannel.close] to unsubscribe from this publisher.
+ * The resulting channel shall be [cancelled][ReceiveChannel.cancel] to unsubscribe from this publisher.
  * @param request how many items to request from publisher in advance (optional, on-demand request by default).
  */
 @JvmOverloads // for binary compatibility
-public fun <T> Publisher<T>.openSubscription(request: Int = 0): SubscriptionReceiveChannel<T> {
+public fun <T> Publisher<T>.openSubscription(request: Int = 0): ReceiveChannel<T> {
     val channel = SubscriptionChannel<T>(request)
     subscribe(channel)
     return channel
@@ -41,7 +41,7 @@ public fun <T> Publisher<T>.openSubscription(request: Int = 0): SubscriptionRece
  */
 @Deprecated(message = "Renamed to `openSubscription`",
     replaceWith = ReplaceWith("openSubscription()"))
-public fun <T> Publisher<T>.open(): SubscriptionReceiveChannel<T> = openSubscription()
+public fun <T> Publisher<T>.open(): ReceiveChannel<T> = openSubscription()
 
 /**
  * Subscribes to this [Publisher] and returns an iterator to receive elements emitted by it.
@@ -60,9 +60,9 @@ public operator fun <T> Publisher<T>.iterator() = openSubscription().iterator()
  * Subscribes to this [Publisher] and performs the specified action for each received element.
  */
 public suspend inline fun <T> Publisher<T>.consumeEach(action: (T) -> Unit) {
-    openSubscription().use { channel ->
-        for (x in channel) action(x)
-    }
+    val channel = openSubscription()
+    for (x in channel) action(x)
+    channel.cancel()
 }
 
 /**
@@ -74,7 +74,7 @@ public suspend fun <T> Publisher<T>.consumeEach(action: suspend (T) -> Unit) =
 
 private class SubscriptionChannel<T>(
     private val request: Int
-) : LinkedListChannel<T>(), SubscriptionReceiveChannel<T>, Subscriber<T> {
+) : LinkedListChannel<T>(), ReceiveChannel<T>, Subscriber<T> {
     init {
         require(request >= 0) { "Invalid request size: $request" }
     }

reactive/kotlinx-coroutines-reactive/src/test/kotlin/kotlinx/coroutines/experimental/reactive/IntegrationTest.kt

@@ -104,7 +104,7 @@ class IntegrationTest(
         checkNumbers(n, pub)
         val channel = pub.openSubscription()
         checkNumbers(n, channel.asPublisher(ctx(coroutineContext)))
-        channel.close()
+        channel.cancel()
     }
 
     private suspend fun checkNumbers(n: Int, pub: Publisher<Int>) {

reactive/kotlinx-coroutines-reactive/src/test/kotlin/kotlinx/coroutines/experimental/reactive/PublisherSubscriptionSelectTest.kt

@@ -40,33 +40,34 @@ class PublisherSubscriptionSelectTest(val request: Int) : TestBase() {
         var a = 0
         var b = 0
         // open two subs
-        source.openSubscription(request).use { channelA ->
-            source.openSubscription(request).use { channelB ->
-                loop@ while (true) {
-                    val done: Int = select {
-                        channelA.onReceiveOrNull {
-                            if (it != null) assertEquals(a++, it)
-                            if (it == null) 0 else 1
-                        }
-                        channelB.onReceiveOrNull {
-                            if (it != null) assertEquals(b++, it)
-                            if (it == null) 0 else 2
-                        }
-                    }
-                    when (done) {
-                        0 -> break@loop
-                        1 -> {
-                            val r = channelB.receiveOrNull()
-                            if (r != null) assertEquals(b++, r)
-                        }
-                        2 -> {
-                            val r = channelA.receiveOrNull()
-                            if (r != null) assertEquals(a++, r)
-                        }
-                    }
+        val channelA = source.openSubscription(request)
+        val channelB = source.openSubscription(request)
+        loop@ while (true) {
+            val done: Int = select {
+                channelA.onReceiveOrNull {
+                    if (it != null) assertEquals(a++, it)
+                    if (it == null) 0 else 1
+                }
+                channelB.onReceiveOrNull {
+                    if (it != null) assertEquals(b++, it)
+                    if (it == null) 0 else 2
+                }
+            }
+            when (done) {
+                0 -> break@loop
+                1 -> {
+                    val r = channelB.receiveOrNull()
+                    if (r != null) assertEquals(b++, r)
+                }
+                2 -> {
+                    val r = channelA.receiveOrNull()
+                    if (r != null) assertEquals(a++, r)
                 }
             }
         }
+
+        channelA.cancel()
+        channelB.cancel()
         // should receive one of them fully
         assertTrue(a == n || b == n)
     }

reactive/kotlinx-coroutines-rx1/src/main/kotlin/kotlinx/coroutines/experimental/rx1/RxChannel.kt

@@ -19,18 +19,18 @@ package kotlinx.coroutines.experimental.rx1
 import kotlinx.atomicfu.atomic
 import kotlinx.atomicfu.loop
 import kotlinx.coroutines.experimental.channels.LinkedListChannel
-import kotlinx.coroutines.experimental.channels.SubscriptionReceiveChannel
+import kotlinx.coroutines.experimental.channels.ReceiveChannel
 import rx.Observable
 import rx.Subscriber
 import rx.Subscription
 
 /**
  * Subscribes to this [Observable] and returns a channel to receive elements emitted by it.
- * The resulting channel shall be [closed][SubscriptionReceiveChannel.close] to unsubscribe from this observable.
+ * The resulting channel shall be [cancelled][ReceiveChannel.cancel] to unsubscribe from this observable.
  * @param request how many items to request from publisher in advance (optional, on-demand request by default).
  */
 @JvmOverloads // for binary compatibility
-public fun <T> Observable<T>.openSubscription(request: Int = 0): SubscriptionReceiveChannel<T> {
+public fun <T> Observable<T>.openSubscription(request: Int = 0): ReceiveChannel<T> {
     val channel = SubscriptionChannel<T>(request)
     val subscription = subscribe(channel.subscriber)
     channel.subscription = subscription
@@ -43,7 +43,7 @@ public fun <T> Observable<T>.openSubscription(request: Int = 0): SubscriptionRec
  */
 @Deprecated(message = "Renamed to `openSubscription`",
     replaceWith = ReplaceWith("openSubscription()"))
-public fun <T> Observable<T>.open(): SubscriptionReceiveChannel<T> = openSubscription()
+public fun <T> Observable<T>.open(): ReceiveChannel<T> = openSubscription()
 
 /**
  * Subscribes to this [Observable] and returns an iterator to receive elements emitted by it.
@@ -62,9 +62,9 @@ public operator fun <T> Observable<T>.iterator() = openSubscription().iterator()
  * Subscribes to this [Observable] and performs the specified action for each received element.
  */
 public suspend inline fun <T> Observable<T>.consumeEach(action: (T) -> Unit) {
-    openSubscription().use { channel ->
-        for (x in channel) action(x)
-    }
+    val channel = openSubscription()
+    for (x in channel) action(x)
+    channel.cancel()
 }
 
 /**
@@ -76,7 +76,7 @@ public suspend fun <T> Observable<T>.consumeEach(action: suspend (T) -> Unit) =
 
 private class SubscriptionChannel<T>(
     private val request: Int
-) : LinkedListChannel<T>(), SubscriptionReceiveChannel<T> {
+) : LinkedListChannel<T>(), ReceiveChannel<T> {
     init {
         require(request >= 0) { "Invalid request size: $request" }
     }

reactive/kotlinx-coroutines-rx1/src/test/kotlin/kotlinx/coroutines/experimental/rx1/IntegrationTest.kt

@@ -126,7 +126,7 @@ class IntegrationTest(
         checkNumbers(n, observable)
         val channel = observable.openSubscription()
         checkNumbers(n, channel.asObservable(ctx(coroutineContext)))
-        channel.close()
+        channel.cancel()
     }
 
     private suspend fun checkNumbers(n: Int, observable: Observable<Int>) {