Repository: arkivanov/Decompose
Pull request #435 (2023-07-25 19:12:56)

Added Value#observe method, deprecated Value#subscribe and Value#unsubscribe methods
Using the deprecated methods may cause memory leaks on iOS.


Patch:
diff --git a/decompose/api/android/decompose.api b/decompose/api/android/decompose.api
index 728a20a2..622b8384 100644
--- a/decompose/api/android/decompose.api
+++ b/decompose/api/android/decompose.api
@@ -5,6 +5,10 @@ public final class com/arkivanov/decompose/BuildConfig {
 	public fun <init> ()V
 }
 
+public abstract interface class com/arkivanov/decompose/Cancellation {
+	public abstract fun cancel ()V
+}
+
 public abstract class com/arkivanov/decompose/Child {
 	public abstract fun getConfiguration ()Ljava/lang/Object;
 	public abstract fun getInstance ()Ljava/lang/Object;
@@ -429,6 +433,7 @@ public final class com/arkivanov/decompose/value/ObserveLifecycleMode : java/lan
 public abstract class com/arkivanov/decompose/value/Value {
 	public fun <init> ()V
 	public abstract fun getValue ()Ljava/lang/Object;
+	public final fun observe (Lkotlin/jvm/functions/Function1;)Lcom/arkivanov/decompose/Cancellation;
 	public abstract fun subscribe (Lkotlin/jvm/functions/Function1;)V
 	public abstract fun unsubscribe (Lkotlin/jvm/functions/Function1;)V
 }
diff --git a/decompose/api/jvm/decompose.api b/decompose/api/jvm/decompose.api
index 7e923224..df3aeaab 100644
--- a/decompose/api/jvm/decompose.api
+++ b/decompose/api/jvm/decompose.api
@@ -1,3 +1,7 @@
+public abstract interface class com/arkivanov/decompose/Cancellation {
+	public abstract fun cancel ()V
+}
+
 public abstract class com/arkivanov/decompose/Child {
 	public abstract fun getConfiguration ()Ljava/lang/Object;
 	public abstract fun getInstance ()Ljava/lang/Object;
@@ -384,6 +388,7 @@ public final class com/arkivanov/decompose/value/ObserveLifecycleMode : java/lan
 public abstract class com/arkivanov/decompose/value/Value {
 	public fun <init> ()V
 	public abstract fun getValue ()Ljava/lang/Object;
+	public final fun observe (Lkotlin/jvm/functions/Function1;)Lcom/arkivanov/decompose/Cancellation;
 	public abstract fun subscribe (Lkotlin/jvm/functions/Function1;)V
 	public abstract fun unsubscribe (Lkotlin/jvm/functions/Function1;)V
 }
diff --git a/decompose/src/commonMain/kotlin/com/arkivanov/decompose/Cancellation.kt b/decompose/src/commonMain/kotlin/com/arkivanov/decompose/Cancellation.kt
new file mode 100644
index 00000000..5c328262
--- /dev/null
+++ b/decompose/src/commonMain/kotlin/com/arkivanov/decompose/Cancellation.kt
@@ -0,0 +1,7 @@
+package com.arkivanov.decompose
+
+/** A cancellation handle, returned by various functions where cancellation is required. */
+fun interface Cancellation {
+
+    fun cancel()
+}
diff --git a/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/MutableValueBuilder.kt b/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/MutableValueBuilder.kt
index bb2fc342..7e2f0dad 100644
--- a/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/MutableValueBuilder.kt
+++ b/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/MutableValueBuilder.kt
@@ -75,6 +75,12 @@ private class MutableValueImpl<T : Any>(initialValue: T) : MutableValue<T>() {
         }
     }
 
+    @Deprecated(
+        "Calling this method from Swift leaks the observer, " +
+            "because Kotlin wraps the function passed from Swift every time the method is called. " +
+            "Please use the new `observe` method which returns `Disposable`.",
+        level = DeprecationLevel.WARNING,
+    )
     override fun subscribe(observer: (T) -> Unit) {
         lock.synchronized {
             if (observer in observers) {
@@ -102,6 +108,12 @@ private class MutableValueImpl<T : Any>(initialValue: T) : MutableValue<T>() {
         }
     }
 
+    @Deprecated(
+        "Calling this method from Swift doesn't have any effect, " +
+            "because Kotlin wraps the function passed from Swift every time the method is called. " +
+            "Please use the new `observe` method which returns `Disposable`.",
+        level = DeprecationLevel.WARNING,
+    )
     override fun unsubscribe(observer: (T) -> Unit) {
         lock.synchronized { observers -= observer }
     }
diff --git a/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/Value.kt b/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/Value.kt
index f9e39b9a..4bc1e933 100644
--- a/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/Value.kt
+++ b/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/Value.kt
@@ -1,5 +1,7 @@
 package com.arkivanov.decompose.value
 
+import com.arkivanov.decompose.Cancellation
+
 /**
  * An observable value holder. Used in Decompose to avoid the direct dependency on coroutines/Reaktive.
  * Also, since [Value] is a class (not an interface) with a generic type parameter, it is useful to
@@ -15,10 +17,36 @@ abstract class Value<out T : Any> {
     /**
      * Subscribes the provided [observer] for value updates. The current value is emitted synchronously on subscription.
      */
+    @Deprecated(
+        message = "Calling this method from Swift leaks the observer, " +
+            "because Kotlin wraps the function passed from Swift every time the method is called. " +
+            "Please use the new `observe` method which returns `Disposable`.",
+        level = DeprecationLevel.WARNING,
+    )
     abstract fun subscribe(observer: (T) -> Unit)
 
     /**
      * Unsubscribes the provided [observer] from value updates.
      */
+    @Deprecated(
+        message = "Calling this method from Swift doesn't have any effect, " +
+            "because Kotlin wraps the function passed from Swift every time the method is called. " +
+            "Please use the new `observe` method which returns `Disposable`.",
+        level = DeprecationLevel.WARNING,
+    )
     abstract fun unsubscribe(observer: (T) -> Unit)
+
+    /**
+     * Subscribes the provided [observer] for value updates. The current value is emitted synchronously on subscription.
+     *
+     * Note: most likely this method will be renamed to `subscribe` in the next major release, once deprecated methods are removed.
+     *
+     * @return [Cancellation] token to cancel the subscription.
+     */
+    @Suppress("DEPRECATION")
+    fun observe(observer: (T) -> Unit): Cancellation {
+        subscribe(observer)
+
+        return Cancellation { unsubscribe(observer) }
+    }
 }
diff --git a/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/ValueExt.kt b/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/ValueExt.kt
index 6e43174a..9542f977 100644
--- a/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/ValueExt.kt
+++ b/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/ValueExt.kt
@@ -1,5 +1,6 @@
 package com.arkivanov.decompose.value
 
+import com.arkivanov.decompose.Cancellation
 import com.arkivanov.essenty.lifecycle.Lifecycle
 import com.arkivanov.essenty.lifecycle.subscribe
 import kotlin.reflect.KProperty
@@ -11,23 +12,25 @@ fun <T : Any> Value<T>.observe(
     mode: ObserveLifecycleMode = ObserveLifecycleMode.START_STOP,
     observer: (T) -> Unit,
 ) {
+    var cancellation: Cancellation? = null
+
     when (mode) {
         ObserveLifecycleMode.CREATE_DESTROY ->
             lifecycle.subscribe(
-                onCreate = { subscribe(observer) },
-                onDestroy = { unsubscribe(observer) }
+                onCreate = { cancellation = observe(observer) },
+                onDestroy = { cancellation?.cancel() },
             )
 
         ObserveLifecycleMode.START_STOP ->
             lifecycle.subscribe(
-                onStart = { subscribe(observer) },
-                onStop = { unsubscribe(observer) }
+                onStart = { cancellation = observe(observer) },
+                onStop = { cancellation?.cancel() },
             )
 
         ObserveLifecycleMode.RESUME_PAUSE ->
             lifecycle.subscribe(
-                onResume = { subscribe(observer) },
-                onPause = { unsubscribe(observer) }
+                onResume = { cancellation = observe(observer) },
+                onPause = { cancellation?.cancel() },
             )
     }.let {}
 }
diff --git a/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/operator/Map.kt b/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/operator/Map.kt
index 8f758ae7..40a76100 100644
--- a/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/operator/Map.kt
+++ b/decompose/src/commonMain/kotlin/com/arkivanov/decompose/value/operator/Map.kt
@@ -28,6 +28,12 @@ private class MappedValue<T : Any, out R : Any>(
             lastMappedValue
         }
 
+    @Deprecated(
+        "Calling this method from Swift leaks the observer, " +
+            "because Kotlin wraps the function passed from Swift every time the method is called. " +
+            "Please use the new `observe` method which returns `Disposable`.",
+        level = DeprecationLevel.WARNING,
+    )
     override fun subscribe(observer: (R) -> Unit) {
         val upstreamObserver: (T) -> Unit = { value -> observer(mapCached(value)) }
 
@@ -39,11 +45,20 @@ private class MappedValue<T : Any, out R : Any>(
             observers[observer] = upstreamObserver
         }
 
+        @Suppress("DEPRECATION")
         upstream.subscribe(upstreamObserver)
     }
 
+    @Deprecated(
+        "Calling this method from Swift doesn't have any effect, " +
+            "because Kotlin wraps the function passed from Swift every time the method is called. " +
+            "Please use the new `observe` method which returns `Disposable`.",
+        level = DeprecationLevel.WARNING,
+    )
     override fun unsubscribe(observer: (R) -> Unit) {
         val upstreamObserver = lock.synchronized { observers.remove(observer) } ?: return
+
+        @Suppress("DEPRECATION")
         upstream.unsubscribe(upstreamObserver)
     }
 }
diff --git a/decompose/src/jsMain/kotlin/com/arkivanov/decompose/router/Utils.kt b/decompose/src/jsMain/kotlin/com/arkivanov/decompose/router/Utils.kt
index b1fc31f3..d6ef3a18 100644
--- a/decompose/src/jsMain/kotlin/com/arkivanov/decompose/router/Utils.kt
+++ b/decompose/src/jsMain/kotlin/com/arkivanov/decompose/router/Utils.kt
@@ -34,7 +34,7 @@ internal fun <T> List<T>.findFirstDifferentIndex(other: List<T>): Int {
 
 internal fun <T : Any> Value<T>.subscribe(observer: (new: T, old: T) -> Unit) {
     var old = value
-    subscribe { new ->
+    observe { new ->
         val tmp = old
         old = new
         observer(new, tmp)
diff --git a/extensions-compose-jetbrains/src/commonMain/kotlin/com/arkivanov/decompose/extensions/compose/jetbrains/SubscribeAsState.kt b/extensions-compose-jetbrains/src/commonMain/kotlin/com/arkivanov/decompose/extensions/compose/jetbrains/SubscribeAsState.kt
index 6921acbe..67246b0b 100644
--- a/extensions-compose-jetbrains/src/commonMain/kotlin/com/arkivanov/decompose/extensions/compose/jetbrains/SubscribeAsState.kt
+++ b/extensions-compose-jetbrains/src/commonMain/kotlin/com/arkivanov/decompose/extensions/compose/jetbrains/SubscribeAsState.kt
@@ -20,13 +20,8 @@ fun <T : Any> Value<T>.subscribeAsState(policy: SnapshotMutationPolicy<T> = stru
     val state = remember(this, policy) { mutableStateOf(value, policy) }
 
     DisposableEffect(this) {
-        val observer: (T) -> Unit = { state.value = it }
-
-        subscribe(observer)
-
-        onDispose {
-            unsubscribe(observer)
-        }
+        val disposable = observe { state.value = it }
+        onDispose { disposable.cancel() }
     }
 
     return state
diff --git a/extensions-compose-jetpack/src/main/java/com/arkivanov/decompose/extensions/compose/jetpack/SubscribeAsState.kt b/extensions-compose-jetpack/src/main/java/com/arkivanov/decompose/extensions/compose/jetpack/SubscribeAsState.kt
index 3acee476..7df81406 100644
--- a/extensions-compose-jetpack/src/main/java/com/arkivanov/decompose/extensions/compose/jetpack/SubscribeAsState.kt
+++ b/extensions-compose-jetpack/src/main/java/com/arkivanov/decompose/extensions/compose/jetpack/SubscribeAsState.kt
@@ -4,17 +4,14 @@ import androidx.compose.runtime.Composable
 import androidx.compose.runtime.DisposableEffect
 import androidx.compose.runtime.SnapshotMutationPolicy
 import androidx.compose.runtime.State
-import androidx.compose.runtime.collectAsState
 import androidx.compose.runtime.mutableStateOf
 import androidx.compose.runtime.remember
 import androidx.compose.runtime.structuralEqualityPolicy
 import com.arkivanov.decompose.value.Value
-import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.StateFlow
 
 /**
  * Subscribes to the [Value] and returns Compose [State].
- * 
+ *
  * @param policy a [SnapshotMutationPolicy] that will be used by Compose when comparing values.
  * Default is [structuralEqualityPolicy].
  */
@@ -23,13 +20,8 @@ fun <T : Any> Value<T>.subscribeAsState(policy: SnapshotMutationPolicy<T> = stru
     val state = remember(this, policy) { mutableStateOf(value, policy) }
 
     DisposableEffect(this) {
-        val observer: (T) -> Unit = { state.value = it }
-
-        subscribe(observer)
-
-        onDispose {
-            unsubscribe(observer)
-        }
+        val cancellation = observe { state.value = it }
+        onDispose { cancellation.cancel() }
     }
 
     return state
diff --git a/sample/app-ios/app-ios/DecomposeHelpers/ObservableValue.swift b/sample/app-ios/app-ios/DecomposeHelpers/ObservableValue.swift
index 7517ad0a..43e68b67 100644
--- a/sample/app-ios/app-ios/DecomposeHelpers/ObservableValue.swift
+++ b/sample/app-ios/app-ios/DecomposeHelpers/ObservableValue.swift
@@ -9,21 +9,17 @@ import SwiftUI
 import Shared
 
 public class ObservableValue<T : AnyObject> : ObservableObject {
-    private let observableValue: Value<T>
-
     @Published
     var value: T
 
-    private var observer: ((T) -> Void)?
+    private var cancellation: Cancellation?
     
     init(_ value: Value<T>) {
-        observableValue = value
-        self.value = observableValue.value
-        observer = { [weak self] value in self?.value = value }
-        observableValue.subscribe(observer: observer!)
+        self.value = value.value
+        self.cancellation = value.observe { [weak self] value in self?.value = value }
     }
 
     deinit {
-        observableValue.unsubscribe(observer: self.observer!)
+        cancellation?.cancel()
     }
 }


Test patch:
diff --git a/decompose/src/commonTest/kotlin/com/arkivanov/decompose/value/MutableValueTest.kt b/decompose/src/commonTest/kotlin/com/arkivanov/decompose/value/MutableValueTest.kt
index b785065f..da8dfc1a 100644
--- a/decompose/src/commonTest/kotlin/com/arkivanov/decompose/value/MutableValueTest.kt
+++ b/decompose/src/commonTest/kotlin/com/arkivanov/decompose/value/MutableValueTest.kt
@@ -27,7 +27,7 @@ class MutableValueTest {
     fun WHEN_subscribe_THEN_current_value_emitted() {
         val values = ArrayList<Int>()
 
-        value.subscribe { values += it }
+        value.observe { values += it }
 
         assertContentEquals(listOf(0), values)
     }
@@ -37,7 +37,7 @@ class MutableValueTest {
         val values = List(10) { ArrayList<Int>() }
 
         repeat(10) { index ->
-            value.subscribe { values[index] += it }
+            value.observe { values[index] += it }
         }
 
         value.value = 1
@@ -50,9 +50,8 @@ class MutableValueTest {
     @Test
     fun GIVEN_unsubscribed_WHEN_value_changed_THEN_not_emitted() {
         val values = ArrayList<Int>()
-        val observer: (Int) -> Unit = { values += it }
-        value.subscribe(observer)
-        value.unsubscribe(observer)
+        val cancellation = value.observe { values += it }
+        cancellation.cancel()
         values.clear()
 
         value.value = 1
@@ -63,10 +62,9 @@ class MutableValueTest {
     @Test
     fun GIVEN_multiple_subscribes_and_one_unsubscribed_WHEN_value_changed_THEN_value_emitted_to_subscribed() {
         val values = ArrayList<Int>()
-        val observer: (Int) -> Unit = {}
-        value.subscribe(observer)
-        value.subscribe { values += it }
-        value.unsubscribe(observer)
+        val cancellation = value.observe {}
+        value.observe { values += it }
+        cancellation.cancel()
         values.clear()
 
         value.value = 1
@@ -77,10 +75,9 @@ class MutableValueTest {
     @Test
     fun GIVEN_multiple_subscribes_and_one_unsubscribed_WHEN_value_changed_THEN_value_not_emitted_to_unsubscribed() {
         val values = ArrayList<Int>()
-        val observer: (Int) -> Unit = { values += it }
-        value.subscribe(observer)
-        value.subscribe {}
-        value.unsubscribe(observer)
+        val cancellation = value.observe { values += it }
+        value.observe {}
+        cancellation.cancel()
         values.clear()
 
         value.value = 1
diff --git a/decompose/src/commonTest/kotlin/com/arkivanov/decompose/value/operator/ValueMapTest.kt b/decompose/src/commonTest/kotlin/com/arkivanov/decompose/value/operator/ValueMapTest.kt
index d8335631..fbda96e8 100644
--- a/decompose/src/commonTest/kotlin/com/arkivanov/decompose/value/operator/ValueMapTest.kt
+++ b/decompose/src/commonTest/kotlin/com/arkivanov/decompose/value/operator/ValueMapTest.kt
@@ -25,7 +25,7 @@ class ValueMapTest {
 
     @Test
     fun GIVEN_subscribed_WHEN_upstream_changed_THEN_value_mapped() {
-        mapped.subscribe {}
+        mapped.observe {}
         upstream.value = "abcd"
 
         val value = mapped.value
@@ -37,7 +37,7 @@ class ValueMapTest {
     fun WHEN_subscribe_THEN_current_value_emitted() {
         val values = ArrayList<Int>()
 
-        mapped.subscribe { values += it }
+        mapped.observe { values += it }
 
         assertContentEquals(listOf(3), values)
     }
@@ -47,7 +47,7 @@ class ValueMapTest {
         val values = List(10) { ArrayList<Int>() }
 
         repeat(10) { index ->
-            mapped.subscribe { values[index] += it }
+            mapped.observe { values[index] += it }
         }
 
         upstream.value = "abcd"
@@ -60,9 +60,8 @@ class ValueMapTest {
     @Test
     fun GIVEN_unsubscribed_WHEN_value_changed_THEN_not_emitted() {
         val values = ArrayList<Int>()
-        val observer: (Int) -> Unit = { values += it }
-        mapped.subscribe(observer)
-        mapped.unsubscribe(observer)
+        val cancellation = mapped.observe { values += it }
+        cancellation.cancel()
         values.clear()
 
         upstream.value = "abcd"
@@ -73,10 +72,9 @@ class ValueMapTest {
     @Test
     fun GIVEN_multiple_subscribes_and_one_unsubscribed_WHEN_value_changed_THEN_value_emitted_to_subscribed() {
         val values = ArrayList<Int>()
-        val observer: (Int) -> Unit = {}
-        mapped.subscribe(observer)
-        mapped.subscribe { values += it }
-        mapped.unsubscribe(observer)
+        val cancellation = mapped.observe {}
+        mapped.observe { values += it }
+        cancellation.cancel()
         values.clear()
 
         upstream.value = "abcd"
@@ -87,10 +85,9 @@ class ValueMapTest {
     @Test
     fun GIVEN_multiple_subscribes_and_one_unsubscribed_WHEN_value_changed_THEN_value_not_emitted_to_unsubscribed() {
         val values = ArrayList<Int>()
-        val observer: (Int) -> Unit = { values += it }
-        mapped.subscribe(observer)
-        mapped.subscribe {}
-        mapped.unsubscribe(observer)
+        val cancellation = mapped.observe { values += it }
+        mapped.observe {}
+        cancellation.cancel()
         values.clear()
 
         upstream.value = "abcd"
@@ -109,9 +106,9 @@ class ValueMapTest {
                     it.length
                 }
 
-        mapped.subscribe {}
-        mapped.subscribe {}
-        mapped.subscribe {}
+        mapped.observe {}
+        mapped.observe {}
+        mapped.observe {}
 
         assertEquals(1, count)
     }
@@ -127,9 +124,9 @@ class ValueMapTest {
                     it.length
                 }
 
-        mapped.subscribe {}
-        mapped.subscribe {}
-        mapped.subscribe {}
+        mapped.observe {}
+        mapped.observe {}
+        mapped.observe {}
         count = 0
 
         upstream.value = "abcd"
@@ -148,11 +145,11 @@ class ValueMapTest {
                     it.length
                 }
 
-        mapped.subscribe {}
-        mapped.subscribe {}
+        mapped.observe {}
+        mapped.observe {}
         upstream.value = "abcd"
         count = 0
-        mapped.subscribe {}
+        mapped.observe {}
 
 
         assertEquals(0, count)
@@ -169,8 +166,8 @@ class ValueMapTest {
                     it.length
                 }
 
-        mapped.subscribe {}
-        mapped.subscribe {}
+        mapped.observe {}
+        mapped.observe {}
         upstream.value = "abcd"
         count = 0
         requireNotNull(upstream.value)
diff --git a/decompose/src/nonJsTest/kotlin/com/arkivanov/decompose/value/AbstractMutableValueThreadingTest.kt b/decompose/src/nonJsTest/kotlin/com/arkivanov/decompose/value/AbstractMutableValueThreadingTest.kt
index 31535a5a..4cad19ea 100644
--- a/decompose/src/nonJsTest/kotlin/com/arkivanov/decompose/value/AbstractMutableValueThreadingTest.kt
+++ b/decompose/src/nonJsTest/kotlin/com/arkivanov/decompose/value/AbstractMutableValueThreadingTest.kt
@@ -16,7 +16,7 @@ abstract class AbstractMutableValueThreadingTest : AbstractThreadingTest() {
             var lastValue = 0
             var counter = 0
 
-            value.subscribe {
+            value.observe {
                 repeat(1000) { counter++ }
                 lastValue = it
                 repeat(1000) { counter-- }
@@ -41,7 +41,7 @@ abstract class AbstractMutableValueThreadingTest : AbstractThreadingTest() {
             var lastValue = 0
             var counter = 0
 
-            value.subscribe {
+            value.observe {
                 repeat(1000) { counter++ }
                 lastValue = it
                 repeat(1000) { counter-- }
@@ -68,7 +68,7 @@ abstract class AbstractMutableValueThreadingTest : AbstractThreadingTest() {
             var counter = 0
 
             repeat(3) {
-                value.subscribe {}
+                value.observe {}
             }
 
             race { threadIndex ->
@@ -76,7 +76,7 @@ abstract class AbstractMutableValueThreadingTest : AbstractThreadingTest() {
                     value.update { it + 1 }
 
                     if ((threadIndex == 0) && (index == iterationCount / 2)) {
-                        value.subscribe {
+                        value.observe {
                             repeat(1000) { counter++ }
                             lastValue = it
                             repeat(1000) { counter-- }
diff --git a/decompose/src/nonJsTest/kotlin/com/arkivanov/decompose/value/operator/AbstractValueMapThreadingTest.kt b/decompose/src/nonJsTest/kotlin/com/arkivanov/decompose/value/operator/AbstractValueMapThreadingTest.kt
index ca287d9d..d80625e3 100644
--- a/decompose/src/nonJsTest/kotlin/com/arkivanov/decompose/value/operator/AbstractValueMapThreadingTest.kt
+++ b/decompose/src/nonJsTest/kotlin/com/arkivanov/decompose/value/operator/AbstractValueMapThreadingTest.kt
@@ -22,7 +22,7 @@ abstract class AbstractValueMapThreadingTest : AbstractThreadingTest() {
             val counters = IntArray(observerCount)
 
             repeat(observerCount) { index ->
-                value.subscribe {
+                value.observe {
                     repeat(1000) { counters[index]++ }
                     lastValues[index] = it
                     repeat(1000) { counters[index]-- }