ValueBroadcastChannel -> ConflatedBroadcastChannel

Author: elizarov

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

@@ -24,10 +24,11 @@ import java.util.concurrent.atomic.AtomicIntegerFieldUpdater
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater
 
 /**
- * Broadcasts the most recently sent value (aka [value]) to all [open] subscribers.
+ * Broadcasts the most recently sent element (aka [value]) to all [open] subscribers.
  *
  * Back-to-send sent elements are _conflated_ -- only the the most recently sent value is received,
  * while previously sent elements **are lost**.
+ * Every subscriber immediately receives the most recently sent element.
  * Sender to this broadcast channel never suspends and [offer] always returns `true`.
  *
  * A secondary constructor can be used to create an instance of this class that already holds a value.
@@ -36,12 +37,12 @@ import java.util.concurrent.atomic.AtomicReferenceFieldUpdater
  * [opening][open] and [closing][SubscriptionReceiveChannel.close] subscription takes O(N) time, where N is the
  * number of subscribers.
  */
-public class ValueBroadcastChannel<E>() : BroadcastChannel<E> {
+public class ConflatedBroadcastChannel<E>() : BroadcastChannel<E> {
     /**
      * Creates an instance of this class that already holds a value.
      *
      * It is as a shortcut to creating an instance with a default constructor and
-     * immediately sending a value: `ValueBroadcastChannel().apply { offer(value) }`.
+     * immediately sending an element: `ConflatedBroadcastChannel().apply { offer(value) }`.
      */
     constructor(value: E) : this() {
         state = State<E>(value, null)
@@ -56,12 +57,12 @@ public class ValueBroadcastChannel<E>() : BroadcastChannel<E> {
 
     private companion object {
         @JvmField
-        val STATE: AtomicReferenceFieldUpdater<ValueBroadcastChannel<*>, Any> = AtomicReferenceFieldUpdater.
-            newUpdater(ValueBroadcastChannel::class.java, Any::class.java, "state")
+        val STATE: AtomicReferenceFieldUpdater<ConflatedBroadcastChannel<*>, Any> = AtomicReferenceFieldUpdater.
+            newUpdater(ConflatedBroadcastChannel::class.java, Any::class.java, "state")
 
         @JvmField
-        val UPDATING: AtomicIntegerFieldUpdater<ValueBroadcastChannel<*>> = AtomicIntegerFieldUpdater.
-            newUpdater(ValueBroadcastChannel::class.java, "updating")
+        val UPDATING: AtomicIntegerFieldUpdater<ConflatedBroadcastChannel<*>> = AtomicIntegerFieldUpdater.
+            newUpdater(ConflatedBroadcastChannel::class.java, "updating")
 
         @JvmField
         val CLOSED = Closed(null)
@@ -257,7 +258,7 @@ public class ValueBroadcastChannel<E>() : BroadcastChannel<E> {
     }
 
     private class Subscriber<E>(
-        private val broadcastChannel: ValueBroadcastChannel<E>
+        private val broadcastChannel: ConflatedBroadcastChannel<E>
     ) : ConflatedChannel<E>(), SubscriptionReceiveChannel<E> {
         override fun close() {
             if (close(cause = null))

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

@@ -23,11 +23,11 @@ import org.hamcrest.core.IsNull
 import org.junit.Assert.*
 import org.junit.Test
 
-class ValueBroadcastChannelTest : TestBase() {
+class ConflatedBroadcastChannelTest : TestBase() {
     @Test
     fun testBasicScenario() = runBlocking {
         expect(1)
-        val broadcast = ValueBroadcastChannel<String>()
+        val broadcast = ConflatedBroadcastChannel<String>()
         assertThat(exceptionFrom { broadcast.value }, IsInstanceOf(IllegalStateException::class.java))
         assertThat(broadcast.valueOrNull, IsNull())
         launch(context, CoroutineStart.UNDISPATCHED) {
@@ -88,7 +88,7 @@ class ValueBroadcastChannelTest : TestBase() {
     @Test
     fun testInitialValueAndReceiveClosed() = runBlocking {
         expect(1)
-        val broadcast = ValueBroadcastChannel<Int>(1)
+        val broadcast = ConflatedBroadcastChannel<Int>(1)
         assertThat(broadcast.value, IsEqual(1))
         assertThat(broadcast.valueOrNull, IsEqual(1))
         launch(context, CoroutineStart.UNDISPATCHED) {

reactive/coroutines-guide-reactive.md

@@ -493,11 +493,11 @@ This is quite the desired behavior for any kind of state-holding variable that n
 other linked state, for example. There is no reason to react to back-to-back updates of the state. 
 Only the most recent state is relevant.
 
-The corresponding behavior in coroutines world is implemented by [ValueBroadcastChannel] that provides the same logic
+The corresponding behavior in coroutines world is implemented by [ConflatedBroadcastChannel] that provides the same logic
 on top of coroutine channels directly, without going through the bridge to the reactive streams:
 
 <!--- INCLUDE
-import kotlinx.coroutines.experimental.channels.ValueBroadcastChannel
+import kotlinx.coroutines.experimental.channels.ConflatedBroadcastChannel
 import kotlinx.coroutines.experimental.channels.consumeEach
 import kotlinx.coroutines.experimental.launch
 import kotlinx.coroutines.experimental.runBlocking
@@ -506,7 +506,7 @@ import kotlinx.coroutines.experimental.yield
 
 ```kotlin
 fun main(args: Array<String>) = runBlocking<Unit> {
-    val broadcast = ValueBroadcastChannel<String>()
+    val broadcast = ConflatedBroadcastChannel<String>()
     broadcast.offer("one")
     broadcast.offer("two")
     // now launch a coroutine to print everything
@@ -1055,7 +1055,7 @@ coroutines for complex pipelines with fan-in and fan-out between multiple worker
 [SubscriptionReceiveChannel.close]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-subscription-receive-channel/close.html
 [SendChannel.send]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-send-channel/send.html
 [BroadcastChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-broadcast-channel/index.html
-[ValueBroadcastChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-value-broadcast-channel/index.html
+[ConflatedBroadcastChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-conflated-broadcast-channel/index.html
 <!--- INDEX kotlinx.coroutines.experimental.selects -->
 [select]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.selects/select.html
 [whileSelect]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.selects/while-select.html

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

@@ -17,14 +17,14 @@
 // This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
 package guide.reactive.basic.example09
 
-import kotlinx.coroutines.experimental.channels.ValueBroadcastChannel
+import kotlinx.coroutines.experimental.channels.ConflatedBroadcastChannel
 import kotlinx.coroutines.experimental.channels.consumeEach
 import kotlinx.coroutines.experimental.launch
 import kotlinx.coroutines.experimental.runBlocking
 import kotlinx.coroutines.experimental.yield
 
 fun main(args: Array<String>) = runBlocking<Unit> {
-    val broadcast = ValueBroadcastChannel<String>()
+    val broadcast = ConflatedBroadcastChannel<String>()
     broadcast.offer("one")
     broadcast.offer("two")
     // now launch a coroutine to print everything