From cc39941ef37779474b28f004dd3dd298b40fcfe6 Mon Sep 17 00:00:00 2001
From: Luki <lukas.vlk@livesport.eu>
Date: Sat, 1 Jun 2024 15:01:26 +0200
Subject: [PATCH] Update ChannelManager.kt (#637)

* Update ChannelManager.kt

Signed-off-by: Luki <luki.sk@gmail.com>

* adding test

Signed-off-by: Luki <luki.sk@gmail.com>

* Update StoreChannelManagerTests.kt

Minor changes for readability, consistency, and scope management

* Update build.gradle.kts

Using Turbine

---------

Signed-off-by: Luki <luki.sk@gmail.com>
Co-authored-by: Matt Ramotar <matt.ramotar@uber.com>
---
 multicast/build.gradle.kts                    | 10 +++
 .../store/multicast5/ChannelManager.kt        |  7 +-
 .../multicast5/StoreChannelManagerTests.kt    | 75 +++++++++++++++++++
 3 files changed, 91 insertions(+), 1 deletion(-)
 create mode 100644 multicast/src/commonTest/kotlin/org/mobilenativefoundation/store/multicast5/StoreChannelManagerTests.kt

diff --git a/multicast/build.gradle.kts b/multicast/build.gradle.kts
index 2b602120f..3d9b1a816 100644
--- a/multicast/build.gradle.kts
+++ b/multicast/build.gradle.kts
@@ -47,6 +47,16 @@ kotlin {
                 implementation(libs.kotlinx.coroutines.core)
             }
         }
+
+        val commonTest by getting {
+            dependencies {
+                implementation(kotlin("test"))
+                implementation(libs.junit)
+                implementation(libs.kotlinx.coroutines.test)
+                implementation(libs.turbine)
+            }
+        }
+
         val jvmMain by getting
         val androidMain by getting
         val nativeMain by creating {
diff --git a/multicast/src/commonMain/kotlin/org/mobilenativefoundation/store/multicast5/ChannelManager.kt b/multicast/src/commonMain/kotlin/org/mobilenativefoundation/store/multicast5/ChannelManager.kt
index 6c35fa005..e0bb23bb6 100644
--- a/multicast/src/commonMain/kotlin/org/mobilenativefoundation/store/multicast5/ChannelManager.kt
+++ b/multicast/src/commonMain/kotlin/org/mobilenativefoundation/store/multicast5/ChannelManager.kt
@@ -22,6 +22,7 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.channels.SendChannel
 import kotlinx.coroutines.flow.Flow
 import org.mobilenativefoundation.store.multicast5.ChannelManager.Message
+import kotlin.coroutines.cancellation.CancellationException
 
 internal interface ChannelManager<T> {
 
@@ -55,7 +56,11 @@ internal interface ChannelManager<T> {
 
         suspend fun dispatchValue(value: Message.Dispatch.Value<T>) {
             _awaitsDispatch = false
-            channel.send(value)
+            try {
+                channel.send(value)
+            } catch (e: CancellationException) {
+                // ignore
+            }
         }
 
         fun dispatchError(error: Throwable) {
diff --git a/multicast/src/commonTest/kotlin/org/mobilenativefoundation/store/multicast5/StoreChannelManagerTests.kt b/multicast/src/commonTest/kotlin/org/mobilenativefoundation/store/multicast5/StoreChannelManagerTests.kt
new file mode 100644
index 000000000..60f4ff859
--- /dev/null
+++ b/multicast/src/commonTest/kotlin/org/mobilenativefoundation/store/multicast5/StoreChannelManagerTests.kt
@@ -0,0 +1,75 @@
+package org.mobilenativefoundation.store.multicast5
+
+
+import app.cash.turbine.test
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.consumeAsFlow
+import kotlinx.coroutines.flow.filterIsInstance
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+class StoreChannelManagerTests {
+
+    @Test
+    fun cancelledDownstreamChannelShouldNotCancelOtherChannels() = runTest {
+        val coroutineScope = CoroutineScope(Dispatchers.Default)
+        val lockUpstream = Mutex(true)
+        val testMessages = listOf(1, 2, 3)
+        val numChannels = 20
+        val upstreamFlow = flow {
+            lockUpstream.withLock {
+                testMessages.onEach { emit(it) }
+            }
+        }
+        val channelManager = StoreChannelManager(
+            scope = coroutineScope,
+            bufferSize = 0,
+            upstream = upstreamFlow,
+            piggybackingDownstream = false,
+            keepUpstreamAlive = false,
+            onEach = { }
+        )
+        val channels = createChannels(numChannels)
+        val channelToBeCancelled = Channel<ChannelManager.Message.Dispatch<Int>>(Channel.UNLIMITED)
+            .also { channel ->
+                coroutineScope.launch {
+                    channel.consumeAsFlow().test {
+                        cancelAndIgnoreRemainingEvents()
+                    }
+                }
+            }
+        coroutineScope.launch {
+            channels.forEach { channelManager.addDownstream(it) }
+            lockUpstream.unlock()
+        }
+        coroutineScope.launch {
+            channelManager.addDownstream(channelToBeCancelled)
+        }
+
+        channels.forEach { channel ->
+            val messagesFlow = channel.consumeAsFlow()
+                .filterIsInstance<ChannelManager.Message.Dispatch.Value<Int>>()
+                .onEach { it.delivered.complete(Unit) }
+
+            messagesFlow.test {
+                for (message in testMessages) {
+                    val dispatchValue = awaitItem()
+                    assertEquals(message, dispatchValue.value)
+                }
+                awaitComplete()
+            }
+        }
+    }
+
+    private fun createChannels(count: Int): List<Channel<ChannelManager.Message.Dispatch<Int>>> {
+        return (1..count).map { Channel(Channel.UNLIMITED) }
+    }
+}