BroadcastChannel.open is renamed to openSubscription

Fixed #54

Author: elizarov

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

@@ -27,7 +27,7 @@ import kotlin.concurrent.withLock
  * Sender suspends only when buffer is full due to one of the receives being slow to consume and
  * receiver suspends only when buffer is empty.
  *
- * Note, that elements that are sent to the broadcast channel while there are no [open] subscribers are immediately
+ * Note, that elements that are sent to the broadcast channel while there are no [openSubscription] subscribers are immediately
  * lost.
  *
  * This implementation uses lock to protect the buffer, which is held only during very short buffer-update operations.
@@ -61,7 +61,7 @@ class ArrayBroadcastChannel<E>(
     override val isBufferAlwaysFull: Boolean get() = false
     override val isBufferFull: Boolean get() = size >= capacity
 
-    override fun open(): SubscriptionReceiveChannel<E> {
+    override fun openSubscription(): SubscriptionReceiveChannel<E> {
         val sub = Subscriber(this, head)
         subs.add(sub)
         // between creating and adding of subscription into the list the buffer head could have been bumped past it,

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

@@ -22,7 +22,7 @@ import java.io.Closeable
 
 /**
  * Broadcast channel is a non-blocking primitive for communication between the sender and multiple receivers
- * that subscribe for the elements using [open] function and unsubscribe using [SubscriptionReceiveChannel.close]
+ * that subscribe for the elements using [openSubscription] function and unsubscribe using [SubscriptionReceiveChannel.close]
  * function.
  *
  * See [BroadcastChannel()][BroadcastChannel.invoke] factory function for the description of available
@@ -55,11 +55,18 @@ public interface BroadcastChannel<E> : SendChannel<E> {
      * The resulting channel shall be [closed][SubscriptionReceiveChannel.close] to unsubscribe from this
      * broadcast channel.
      */
-    public fun open(): SubscriptionReceiveChannel<E>
+    public fun openSubscription(): SubscriptionReceiveChannel<E>
+
+    /**
+     * @suppress **Deprecated**: Renamed to [openSubscription]
+     */
+    @Deprecated(message = "Renamed to `openSubscription`",
+        replaceWith = ReplaceWith("openSubscription()"))
+    public fun open(): SubscriptionReceiveChannel<E> = openSubscription()
 }
 
 /**
- * Return type for [BroadcastChannel.open] that can be used to [receive] elements from the
+ * Return type for [BroadcastChannel.openSubscription] that can be used to [receive] elements from the
  * open subscription and to [close] it to unsubscribe.
  */
 public interface SubscriptionReceiveChannel<out T> : ReceiveChannel<T>, Closeable {

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

@@ -31,7 +31,7 @@ public suspend fun <E> ReceiveChannel<E>.consumeEach(action: suspend (E) -> Unit
  */
 // :todo: make it inline when this bug is fixed: https://youtrack.jetbrains.com/issue/KT-16448
 public suspend fun <E> BroadcastChannel<E>.consumeEach(action: suspend (E) -> Unit) {
-    open().use { channel ->
+    openSubscription().use { channel ->
         for (x in channel) action(x)
     }
 }

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

@@ -23,7 +23,7 @@ import java.util.concurrent.atomic.AtomicIntegerFieldUpdater
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater
 
 /**
- * Broadcasts the most recently sent element (aka [value]) to all [open] subscribers.
+ * Broadcasts the most recently sent element (aka [value]) to all [openSubscription] subscribers.
  *
  * Back-to-send sent elements are _conflated_ -- only the the most recently sent value is received,
  * while previously sent elements **are lost**.
@@ -33,7 +33,7 @@ import java.util.concurrent.atomic.AtomicReferenceFieldUpdater
  * A secondary constructor can be used to create an instance of this class that already holds a value.
  *
  * This implementation is fully lock-free. In this implementation
- * [opening][open] and [closing][SubscriptionReceiveChannel.close] subscription takes O(N) time, where N is the
+ * [opening][openSubscription] and [closing][SubscriptionReceiveChannel.close] subscription takes O(N) time, where N is the
  * number of subscribers.
  */
 public class ConflatedBroadcastChannel<E>() : BroadcastChannel<E> {
@@ -124,7 +124,7 @@ public class ConflatedBroadcastChannel<E>() : BroadcastChannel<E> {
     override val isFull: Boolean get() = false
 
     @Suppress("UNCHECKED_CAST")
-    override fun open(): SubscriptionReceiveChannel<E> {
+    override fun openSubscription(): SubscriptionReceiveChannel<E> {
         val subscriber = Subscriber<E>(this)
         while (true) { // lock-free loop on state
             val state = this.state

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

@@ -28,7 +28,7 @@ class ArrayBroadcastChannelTest : TestBase() {
         expect(1)
         val broadcast = ArrayBroadcastChannel<Int>(1)
         assertThat(broadcast.isClosedForSend, IsEqual(false))
-        val first = broadcast.open()
+        val first = broadcast.openSubscription()
         launch(context, CoroutineStart.UNDISPATCHED) {
             expect(2)
             assertThat(first.receive(), IsEqual(1)) // suspends
@@ -46,7 +46,7 @@ class ArrayBroadcastChannelTest : TestBase() {
         expect(4)
         yield() // to the first receiver
         expect(6)
-        val second = broadcast.open()
+        val second = broadcast.openSubscription()
         launch(context, CoroutineStart.UNDISPATCHED) {
             expect(7)
             assertThat(second.receive(), IsEqual(2)) // suspends
@@ -72,7 +72,7 @@ class ArrayBroadcastChannelTest : TestBase() {
     fun testSendSuspend() = runBlocking {
         expect(1)
         val broadcast = ArrayBroadcastChannel<Int>(1)
-        val first = broadcast.open()
+        val first = broadcast.openSubscription()
         launch(context) {
             expect(4)
             assertThat(first.receive(), IsEqual(1))
@@ -91,7 +91,7 @@ class ArrayBroadcastChannelTest : TestBase() {
     fun testConcurrentSendCompletion() = runBlocking {
         expect(1)
         val broadcast = ArrayBroadcastChannel<Int>(1)
-        val sub = broadcast.open()
+        val sub = broadcast.openSubscription()
         // launch 3 concurrent senders (one goes buffer, two other suspend)
         for (x in 1..3) {
             launch(context, CoroutineStart.UNDISPATCHED) {
@@ -121,7 +121,7 @@ class ArrayBroadcastChannelTest : TestBase() {
         broadcast.send(2)
         broadcast.send(3)
         expect(2) // should not suspend anywhere above
-        val sub = broadcast.open()
+        val sub = broadcast.openSubscription()
         launch(context, CoroutineStart.UNDISPATCHED) {
             expect(3)
             assertThat(sub.receive(), IsEqual(4)) // suspends

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

@@ -93,7 +93,7 @@ class ConflatedBroadcastChannelNotifyStressTest : TestBase() {
     }
 
     suspend fun waitForEvent(): Int =
-        broadcast.open().use {
+        broadcast.openSubscription().use {
             it.receive()
         }
 }

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

@@ -20,7 +20,7 @@ import kotlinx.coroutines.experimental.*
 import org.hamcrest.core.IsEqual
 import org.hamcrest.core.IsInstanceOf
 import org.hamcrest.core.IsNull
-import org.junit.Assert.*
+import org.junit.Assert.assertThat
 import org.junit.Test
 
 class ConflatedBroadcastChannelTest : TestBase() {
@@ -32,7 +32,7 @@ class ConflatedBroadcastChannelTest : TestBase() {
         assertThat(broadcast.valueOrNull, IsNull())
         launch(context, CoroutineStart.UNDISPATCHED) {
             expect(2)
-            val sub = broadcast.open()
+            val sub = broadcast.openSubscription()
             assertThat(sub.poll(), IsNull())
             expect(3)
             assertThat(sub.receive(), IsEqual("one")) // suspends
@@ -51,7 +51,7 @@ class ConflatedBroadcastChannelTest : TestBase() {
         expect(7)
         launch(context, CoroutineStart.UNDISPATCHED) {
             expect(8)
-            val sub = broadcast.open()
+            val sub = broadcast.openSubscription()
             assertThat(sub.receive(), IsEqual("one")) // does not suspend
             expect(9)
             assertThat(sub.receive(), IsEqual("two")) // suspends
@@ -93,7 +93,7 @@ class ConflatedBroadcastChannelTest : TestBase() {
         assertThat(broadcast.valueOrNull, IsEqual(1))
         launch(context, CoroutineStart.UNDISPATCHED) {
             expect(2)
-            val sub = broadcast.open()
+            val sub = broadcast.openSubscription()
             assertThat(sub.receive(), IsEqual(1))
             expect(3)
             assertThat(exceptionFrom { sub.receive() }, IsInstanceOf(ClosedReceiveChannelException::class.java)) // suspends

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

@@ -62,7 +62,7 @@ enum class TestChannelKind {
 private class ChannelViaBroadcast<E>(
     private val broadcast: BroadcastChannel<E>
 ): Channel<E>, SendChannel<E> by broadcast {
-    val sub = broadcast.open()
+    val sub = broadcast.openSubscription()
 
     override val isClosedForReceive: Boolean get() = sub.isClosedForReceive
     override val isEmpty: Boolean get() = sub.isEmpty

reactive/coroutines-guide-reactive.md

@@ -223,7 +223,7 @@ method with it.
 
 An example in the previous section uses `source.consumeEach { ... }` snippet to open a subscription 
 and receive all the elements from it. If we need more control on how what to do with 
-the elements that are being received from the channel, we can use [Publisher.open][org.reactivestreams.Publisher.open]
+the elements that are being received from the channel, we can use [Publisher.openSubscription][org.reactivestreams.Publisher.openSubscription]
 as shown in the following example:
 
 <!--- INCLUDE
@@ -238,7 +238,7 @@ 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.open().use { channel -> // open channel to the source
+    source.openSubscription().use { channel -> // 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
@@ -262,7 +262,7 @@ Finally
 
 <!--- TEST -->
 
-With an explicit `open` we should [close][SubscriptionReceiveChannel.close] the corresponding 
+With an explicit `openSubscription` we should [close][SubscriptionReceiveChannel.close] the corresponding 
 subscription to unsubscribe from the source. However, instead of invoking `close` explicitly, 
 this code relies on [use](https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.io/use.html)
 function from Kotlin's standard library.
@@ -676,11 +676,11 @@ import kotlinx.coroutines.experimental.selects.whileSelect
 
 ```kotlin
 fun <T, U> Publisher<T>.takeUntil(context: CoroutineContext, other: Publisher<U>) = publish<T>(context) {
-    this@takeUntil.open().use { thisChannel ->           // explicitly open channel to Publisher<T>
-        other.open().use { otherChannel ->               // explicitly open channel to Publisher<U>
+    this@takeUntil.openSubscription().use { thisChannel -> // explicitly open channel to Publisher<T>
+        other.openSubscription().use { otherChannel ->     // 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
+                otherChannel.onReceive { false }          // bail out on any received element from `other`
+                thisChannel.onReceive { send(it); true }  // resend element from this channel and continue
             }
         }
     }
@@ -1071,7 +1071,7 @@ coroutines for complex pipelines with fan-in and fan-out between multiple worker
 <!--- INDEX kotlinx.coroutines.experimental.reactive -->
 [publish]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.experimental.reactive/publish.html
 [org.reactivestreams.Publisher.consumeEach]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.experimental.reactive/org.reactivestreams.-publisher/consume-each.html
-[org.reactivestreams.Publisher.open]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.experimental.reactive/org.reactivestreams.-publisher/open.html
+[org.reactivestreams.Publisher.openSubscription]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.experimental.reactive/org.reactivestreams.-publisher/open-subscription.html
 <!--- MODULE kotlinx-coroutines-rx2 -->
 <!--- INDEX kotlinx.coroutines.experimental.rx2 -->
 [rxFlowable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.experimental.rx2/rx-flowable.html

reactive/kotlinx-coroutines-reactive/README.md

@@ -16,7 +16,7 @@ Suspending extension functions and suspending iteration:
 | [Publisher.awaitFirstOrDefault][org.reactivestreams.Publisher.awaitFirstOrDefault] | Returns the first value from the given publisher or default
 | [Publisher.awaitLast][org.reactivestreams.Publisher.awaitFirst] | Returns the last value from the given publisher
 | [Publisher.awaitSingle][org.reactivestreams.Publisher.awaitSingle] | Returns the single value from the given publisher
-| [Publisher.open][org.reactivestreams.Publisher.open] | Subscribes to publisher and returns [ReceiveChannel] 
+| [Publisher.openSubscription][org.reactivestreams.Publisher.openSubscription] | Subscribes to publisher and returns [ReceiveChannel] 
 | [Publisher.iterator][org.reactivestreams.Publisher.iterator] | Subscribes to publisher and returns [ChannelIterator]
 
 Conversion functions:
@@ -37,7 +37,7 @@ Conversion functions:
 [org.reactivestreams.Publisher.awaitFirst]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.experimental.reactive/org.reactivestreams.-publisher/await-first.html
 [org.reactivestreams.Publisher.awaitFirstOrDefault]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.experimental.reactive/org.reactivestreams.-publisher/await-first-or-default.html
 [org.reactivestreams.Publisher.awaitSingle]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.experimental.reactive/org.reactivestreams.-publisher/await-single.html
-[org.reactivestreams.Publisher.open]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.experimental.reactive/org.reactivestreams.-publisher/open.html
+[org.reactivestreams.Publisher.openSubscription]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.experimental.reactive/org.reactivestreams.-publisher/open-subscription.html
 [org.reactivestreams.Publisher.iterator]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.experimental.reactive/org.reactivestreams.-publisher/iterator.html
 [kotlinx.coroutines.experimental.channels.ReceiveChannel.asPublisher]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.experimental.reactive/kotlinx.coroutines.experimental.channels.-receive-channel/as-publisher.html
 <!--- END -->