Add Subscribable interfaces

common/kotlinx-coroutines-core-common/src/channels/BroadcastChannel.kt

@@ -18,7 +18,7 @@ import kotlinx.coroutines.experimental.internalAnnotations.*
  * See `BroadcastChannel()` factory function for the description of available
  * broadcast channel implementations.
  */
-public interface BroadcastChannel<E> : SendChannel<E> {
+public interface BroadcastChannel<E> : SendChannel<E>, Subscribable<E> {
     /**
      * Factory for broadcast channels.
      * @suppress **Deprecated**
@@ -37,7 +37,8 @@ public interface BroadcastChannel<E> : SendChannel<E> {
      * The resulting channel shall be [cancelled][ReceiveChannel.cancel] to unsubscribe from this
      * broadcast channel.
      */
-    public fun openSubscription(): ReceiveChannel<E>
+    // TODO: Remove (already declared in Subscribable). Kept here only for binary compatibility
+    public override fun openSubscription(): ReceiveChannel<E>
 
     /**
      * @suppress **Deprecated**: Return type changed to `ReceiveChannel`, this one left here for binary compatibility.

common/kotlinx-coroutines-core-common/src/channels/Subscribables.kt

@@ -0,0 +1,104 @@
+/*
+ * Copyright 2016-2018 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
+
+import kotlinx.coroutines.experimental.Unconfined
+import kotlin.properties.ReadOnlyProperty
+import kotlin.properties.ReadWriteProperty
+import kotlin.reflect.KProperty
+
+/**
+ * Interface for classes allowing to open subscriptions.
+ */
+public interface Subscribable<out T> {
+
+    /**
+     * Subscribes and returns a channel to receive elements broadcasted by this Subscribable.
+     * The resulting channel shall be [cancelled][ReceiveChannel.cancel] to unsubscribe from this subscribable.
+     */
+    public fun openSubscription(): ReceiveChannel<T>
+}
+
+/**
+ * Interface for a value holder where the value may change over time.
+ *
+ * The current value can be obtained with [value] and one can open a subscription in order to receive the new value when changed.
+ *
+ * The channel returned by [openSubscription] is "conflated" and start by the current value if any.
+ */
+public interface SubscribableValue<out T> : Subscribable<T>, ReadOnlyProperty<Any?, T> {
+
+    /**
+     * Current value
+     */
+    public val value: T
+
+    /**
+     * Returns the current value.
+     *
+     * @see value
+     */
+    public override fun getValue(thisRef: Any?, property: KProperty<*>): T = value
+}
+
+/**
+ * Mutable version of [SubscribableValue].
+ *
+ * Provide ability to get and set the current value with [value].
+ */
+public interface SubscribableVariable<T> : SubscribableValue<T>, ReadWriteProperty<Any?, T> {
+
+    /**
+     * Current value
+     */
+    public override var value: T
+
+    public override fun getValue(thisRef: Any?, property: KProperty<*>): T = value
+
+    /**
+     * Set the current value
+     */
+    public override fun setValue(thisRef: Any?, property: KProperty<*>, value: T) {
+        this.value = value
+    }
+}
+
+/**
+ * Create an instance of [SubscribableValue] with the given [value]
+ */
+public fun <T> SubscribableValue(value: T): SubscribableValue<T> = SubscribableValueImpl(value)
+
+/**
+ * Create an instance of [SubscribableVariable] initialized with the given [initialValue]
+ */
+public fun <T> SubscribableVariable(initialValue: T): SubscribableVariable<T> = SubscribableVariableImpl(initialValue)
+
+private class SubscribableValueImpl<T>(override val value: T): SubscribableValue<T> {
+    override fun openSubscription() = produce(Unconfined) { send(value) }
+}
+
+private class SubscribableVariableImpl<T>(initialValue: T): SubscribableVariable<T> {
+    private val broadcast = ConflatedBroadcastChannel(initialValue)
+
+    override var value: T
+        get() = broadcast.value
+        set(value) {
+            broadcast.offer(value)
+        }
+
+    override fun openSubscription() = broadcast.openSubscription()
+}

common/kotlinx-coroutines-core-common/test/channels/SubscribableValueTest.kt

@@ -0,0 +1,63 @@
+package kotlinx.coroutines.experimental.channels
+
+import kotlinx.coroutines.experimental.CoroutineStart
+import kotlinx.coroutines.experimental.TestBase
+import kotlinx.coroutines.experimental.Unconfined
+import kotlinx.coroutines.experimental.launch
+import kotlin.coroutines.experimental.coroutineContext
+import kotlin.test.Test
+
+class SubscribableValueTest : TestBase() {
+
+    @Test
+    fun testBasicScenario() = runTest {
+        expect(1)
+        val subscribable = SubscribableValue(42)
+        val variable by subscribable
+        check(subscribable.value == 42)
+        check(variable == 42)
+        launch(coroutineContext, start = CoroutineStart.UNDISPATCHED) {
+            expect(2)
+            val sub = subscribable.openSubscription()
+            check(sub.receive() == 42)
+            expect(3)
+            try {
+                sub.receive()
+                expectUnreached()
+            } catch (e: ClosedReceiveChannelException) {
+                // expected exception
+            }
+            expect(4)
+        }
+        finish(5)
+    }
+
+    @Test
+    fun testOpenSubscriptionAlwaysStartByTheCurrentValue() {
+        expect(1)
+        val subscribable = SubscribableValue(42)
+        launch(Unconfined) {
+            expect(2)
+            check(subscribable.openSubscription().first() == 42)
+            check(subscribable.openSubscription().first() == 42)
+            expect(3)
+        }
+        finish(4)
+    }
+
+    @Test
+    fun testObservableValue() {
+        expect(1)
+        val subscribable = SubscribableValue(42)
+        val value by subscribable
+        check(subscribable.value == 42)
+        check(value == 42)
+        launch(Unconfined) {
+            expect(2)
+            check(subscribable.openSubscription().first() == 42)
+            check(subscribable.openSubscription().first() == 42)
+            expect(3)
+        }
+        finish(4)
+    }
+}

common/kotlinx-coroutines-core-common/test/channels/SubscribableVariableTest.kt

@@ -0,0 +1,65 @@
+package kotlinx.coroutines.experimental.channels
+
+import kotlinx.coroutines.experimental.CoroutineStart
+import kotlinx.coroutines.experimental.TestBase
+import kotlinx.coroutines.experimental.Unconfined
+import kotlinx.coroutines.experimental.launch
+import kotlinx.coroutines.experimental.yield
+import kotlin.coroutines.experimental.coroutineContext
+import kotlin.test.Test
+
+class SubscribableVariableTest : TestBase() {
+
+    @kotlin.test.Test
+    fun testBasicScenario() = runTest {
+        expect(1)
+        val subscribable = SubscribableVariable(42)
+        var variable by subscribable
+        check(subscribable.value == 42)
+        check(variable == 42)
+        launch(coroutineContext, start = CoroutineStart.UNDISPATCHED) {
+            expect(2)
+            val sub = subscribable.openSubscription()
+            check(sub.receive() == 42)
+            expect(3)
+            check(sub.receive() == 24) // suspends
+            expect(6)
+        }
+        expect(4)
+        variable = 24
+        check(subscribable.value == 24)
+        check(variable == 24)
+        expect(5)
+        yield() // to child
+        finish(7)
+    }
+
+    @Test
+    fun testOpenSubscriptionAlwaysStartByTheCurrentValue() {
+        expect(1)
+        val subscribable = SubscribableVariable(42)
+        launch(Unconfined) {
+            expect(2)
+            check(subscribable.openSubscription().first() == 42)
+            check(subscribable.openSubscription().first() == 42)
+            expect(3)
+        }
+        finish(4)
+    }
+
+    @Test
+    fun testObservableValue() {
+        expect(1)
+        val subscribable = SubscribableValue(42)
+        val value by subscribable
+        check(subscribable.value == 42)
+        check(value == 42)
+        launch(Unconfined) {
+            expect(2)
+            check(subscribable.openSubscription().first() == 42)
+            check(subscribable.openSubscription().first() == 42)
+            expect(3)
+        }
+        finish(4)
+    }
+}