From d09361b0f75b56dd43f16f1eb67bbf5e73b5b89c Mon Sep 17 00:00:00 2001
From: Arkadii Ivanov <arkann1985@gmail.com>
Date: Thu, 8 Aug 2024 13:33:57 +0100
Subject: [PATCH] Added Store#labelsChannel API

---
 .../extensions/coroutines/StoreExt.kt         |  42 ++++++-
 .../extensions/coroutines/LabelChannelTest.kt | 104 ++++++++++++++++++
 .../extensions/coroutines/LabelFlowTest.kt    |  77 +++++++++++++
 3 files changed, 222 insertions(+), 1 deletion(-)
 create mode 100644 mvikotlin-extensions-coroutines/src/commonTest/kotlin/com/arkivanov/mvikotlin/extensions/coroutines/LabelChannelTest.kt
 create mode 100644 mvikotlin-extensions-coroutines/src/commonTest/kotlin/com/arkivanov/mvikotlin/extensions/coroutines/LabelFlowTest.kt

diff --git a/mvikotlin-extensions-coroutines/src/commonMain/kotlin/com/arkivanov/mvikotlin/extensions/coroutines/StoreExt.kt b/mvikotlin-extensions-coroutines/src/commonMain/kotlin/com/arkivanov/mvikotlin/extensions/coroutines/StoreExt.kt
index 744869f2..52a99167 100644
--- a/mvikotlin-extensions-coroutines/src/commonMain/kotlin/com/arkivanov/mvikotlin/extensions/coroutines/StoreExt.kt
+++ b/mvikotlin-extensions-coroutines/src/commonMain/kotlin/com/arkivanov/mvikotlin/extensions/coroutines/StoreExt.kt
@@ -2,14 +2,19 @@ package com.arkivanov.mvikotlin.extensions.coroutines
 
 import com.arkivanov.mvikotlin.core.rx.observer
 import com.arkivanov.mvikotlin.core.store.Store
+import com.arkivanov.mvikotlin.core.utils.ExperimentalMviKotlinApi
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.awaitCancellation
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.FlowCollector
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.stateIn
+import kotlinx.coroutines.launch
 import kotlin.coroutines.CoroutineContext
 
 /**
@@ -69,7 +74,42 @@ private class StoreStateFlow<out State : Any>(
 /**
  * Returns a [Flow] that emits [Store] labels.
  *
- *  Please note that the actual collection of the [Flow] may not be synchronous depending on [CoroutineContext] being used.
+ * Please note that the actual collection of the [Flow] may not be synchronous depending on [CoroutineContext] being used.
  */
 val <Label : Any> Store<*, *, Label>.labels: Flow<Label>
     get() = toFlow(Store<*, *, Label>::labels)
+
+/**
+ * Returns a [ReceiveChannel] that emits [Store] labels. Unlike [labels] that returns a [Flow], this API
+ * is useful when labels must not be skipped while there is no subscriber. Please keep in mind that labels
+ * still may be skipped if they are dispatched synchronously on [Store] initialization. If that's the case,
+ * you can disable the automatic initialization by passing `autoInit = false` parameter when creating a [Store],
+ * see [StoreFactory.create][com.arkivanov.mvikotlin.core.store.StoreFactory.create] for more information.
+ *
+ * Due to the nature of how channels work, it is recommended to have one [Channel] per subscriber.
+ *
+ * Please note that the actual collection of the [Flow] may not be synchronous depending on [CoroutineContext] being used.
+ *
+ * @param scope a [CoroutineScope] used for cancelling the underlying [Channel].
+ * @param capacity a capacity of the underlying [Channel], default value is [Channel.BUFFERED].
+ */
+@ExperimentalMviKotlinApi
+fun <Label : Any> Store<*, *, Label>.labelsChannel(
+    scope: CoroutineScope,
+    capacity: Int = Channel.BUFFERED,
+): ReceiveChannel<Label> {
+    val channel = Channel<Label>(capacity = capacity)
+    val disposable = labels(observer(onNext = channel::trySend))
+
+    scope.launch {
+        try {
+            awaitCancellation()
+        } finally {
+            disposable.dispose()
+            channel.cancel()
+        }
+    }
+
+    return channel
+}
+
diff --git a/mvikotlin-extensions-coroutines/src/commonTest/kotlin/com/arkivanov/mvikotlin/extensions/coroutines/LabelChannelTest.kt b/mvikotlin-extensions-coroutines/src/commonTest/kotlin/com/arkivanov/mvikotlin/extensions/coroutines/LabelChannelTest.kt
new file mode 100644
index 00000000..6411584a
--- /dev/null
+++ b/mvikotlin-extensions-coroutines/src/commonTest/kotlin/com/arkivanov/mvikotlin/extensions/coroutines/LabelChannelTest.kt
@@ -0,0 +1,104 @@
+package com.arkivanov.mvikotlin.extensions.coroutines
+
+import com.arkivanov.mvikotlin.core.rx.Disposable
+import com.arkivanov.mvikotlin.core.rx.Observer
+import com.arkivanov.mvikotlin.core.store.Store
+import com.arkivanov.mvikotlin.core.utils.ExperimentalMviKotlinApi
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.launch
+import kotlin.test.Test
+import kotlin.test.assertContentEquals
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+@OptIn(ExperimentalMviKotlinApi::class)
+@Suppress("TestFunctionName")
+class LabelChannelTest {
+
+    @Test
+    fun WHEN_label_emitted_THEN_label_collected() {
+        val store = TestStore()
+        val scope = CoroutineScope(Dispatchers.Unconfined)
+        val channel = store.labelsChannel(scope)
+        val labels = ArrayList<Int>()
+
+        store.labelObserver?.onNext(1)
+
+        scope.launch {
+            for (label in channel) {
+                labels += label
+            }
+        }
+
+        store.labelObserver?.onNext(2)
+        store.labelObserver?.onNext(3)
+
+        assertContentEquals(listOf(1, 2, 3), labels)
+    }
+
+    @Test
+    fun WHEN_scope_cancelled_THEN_unsubscribed_from_store() {
+        val store = TestStore()
+        val scope = CoroutineScope(Dispatchers.Unconfined)
+        val channel = store.labelsChannel(scope)
+
+        scope.launch {
+            while (true) {
+                channel.receive()
+            }
+        }
+
+        scope.cancel()
+
+        assertNull(store.labelObserver)
+    }
+
+    @OptIn(DelicateCoroutinesApi::class)
+    @Test
+    fun WHEN_scope_cancelled_THEN_channel_cancelled() {
+        val store = TestStore()
+        val scope = CoroutineScope(Dispatchers.Unconfined)
+        val channel = store.labelsChannel(scope)
+
+        scope.launch {
+            while (true) {
+                channel.receive()
+            }
+        }
+
+        scope.cancel()
+
+        assertTrue(channel.isClosedForReceive)
+    }
+
+    private class TestStore : Store<Int, Int, Int> {
+        override val state: Int = 0
+        override val isDisposed: Boolean = false
+
+        var labelObserver: Observer<Int>? = null
+            private set
+
+        override fun states(observer: Observer<Int>): Disposable = error("Not required")
+
+        override fun labels(observer: Observer<Int>): Disposable {
+            labelObserver = observer
+
+            return Disposable { labelObserver = null }
+        }
+
+        override fun accept(intent: Int) {
+            // no-op
+        }
+
+        override fun init() {
+            // no-op
+        }
+
+        override fun dispose() {
+            // no-op
+        }
+    }
+}
diff --git a/mvikotlin-extensions-coroutines/src/commonTest/kotlin/com/arkivanov/mvikotlin/extensions/coroutines/LabelFlowTest.kt b/mvikotlin-extensions-coroutines/src/commonTest/kotlin/com/arkivanov/mvikotlin/extensions/coroutines/LabelFlowTest.kt
new file mode 100644
index 00000000..b9d35daa
--- /dev/null
+++ b/mvikotlin-extensions-coroutines/src/commonTest/kotlin/com/arkivanov/mvikotlin/extensions/coroutines/LabelFlowTest.kt
@@ -0,0 +1,77 @@
+package com.arkivanov.mvikotlin.extensions.coroutines
+
+import com.arkivanov.mvikotlin.core.rx.Disposable
+import com.arkivanov.mvikotlin.core.rx.Observer
+import com.arkivanov.mvikotlin.core.store.Store
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.launch
+import kotlin.test.Test
+import kotlin.test.assertContentEquals
+import kotlin.test.assertNull
+
+@Suppress("TestFunctionName")
+class LabelFlowTest {
+
+    @Test
+    fun WHEN_label_emitted_THEN_label_collected() {
+        val store = TestStore()
+        val flow = store.labels
+        val scope = CoroutineScope(Dispatchers.Unconfined)
+        val items = ArrayList<Int>()
+
+        scope.launch {
+            flow.collect { items += it }
+        }
+
+        store.labelObserver?.onNext(1)
+        store.labelObserver?.onNext(2)
+        store.labelObserver?.onNext(3)
+
+        assertContentEquals(listOf(1, 2, 3), items)
+    }
+
+    @Test
+    fun WHEN_collection_cancelled_THEN_unsubscribed_from_store() {
+        val store = TestStore()
+        val flow = store.labels
+        val scope = CoroutineScope(Dispatchers.Unconfined)
+
+        scope.launch {
+            flow.collect {}
+        }
+
+        scope.cancel()
+
+        assertNull(store.labelObserver)
+    }
+
+    private class TestStore : Store<Int, Int, Int> {
+        override val state: Int = 0
+        override val isDisposed: Boolean = false
+
+        var labelObserver: Observer<Int>? = null
+            private set
+
+        override fun states(observer: Observer<Int>): Disposable = error("Not required")
+
+        override fun labels(observer: Observer<Int>): Disposable {
+            labelObserver = observer
+
+            return Disposable { labelObserver = null }
+        }
+
+        override fun accept(intent: Int) {
+            // no-op
+        }
+
+        override fun init() {
+            // no-op
+        }
+
+        override fun dispose() {
+            // no-op
+        }
+    }
+}