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 61948bd93..aa913f577 100644 --- a/multicast/src/commonMain/kotlin/org/mobilenativefoundation/store/multicast5/ChannelManager.kt +++ b/multicast/src/commonMain/kotlin/org/mobilenativefoundation/store/multicast5/ChannelManager.kt @@ -321,6 +321,12 @@ internal class StoreChannelManager( channels.removeAt(index) if (!keepUpstreamAlive && channels.isEmpty()) { producer?.cancelAndJoin() + // Clear the dead producer reference right away instead of waiting for its + // UpstreamFinished message. Otherwise a downstream added before that message + // arrives would not restart the upstream and would never receive a value. + // The stale UpstreamFinished is ignored by the identity check in + // doHandleUpstreamClose. + producer = null } } } diff --git a/multicast/src/commonTest/kotlin/org/mobilenativefoundation/store/multicast5/StoreChannelManagerTests.kt b/multicast/src/commonTest/kotlin/org/mobilenativefoundation/store/multicast5/StoreChannelManagerTests.kt index c765b7c11..327693b6e 100644 --- a/multicast/src/commonTest/kotlin/org/mobilenativefoundation/store/multicast5/StoreChannelManagerTests.kt +++ b/multicast/src/commonTest/kotlin/org/mobilenativefoundation/store/multicast5/StoreChannelManagerTests.kt @@ -3,6 +3,8 @@ package org.mobilenativefoundation.store.multicast5 import app.cash.turbine.test import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.flow.consumeAsFlow import kotlinx.coroutines.flow.filterIsInstance @@ -11,10 +13,12 @@ import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.test.advanceUntilIdle import kotlinx.coroutines.test.runTest import kotlin.test.Test import kotlin.test.assertEquals +@OptIn(ExperimentalCoroutinesApi::class) class StoreChannelManagerTests { @Test fun cancelledDownstreamChannelShouldNotCancelOtherChannels() = @@ -72,6 +76,48 @@ class StoreChannelManagerTests { } } + @Test + fun downstreamAddedWhileUpstreamCancellationIsInFlightShouldRestartUpstream() = + runTest { + var upstreamCollectionCount = 0 + val upstreamFlow = + flow { + upstreamCollectionCount++ + if (upstreamCollectionCount == 1) { + awaitCancellation() + } else { + emit(1) + } + } + val channelManager = + StoreChannelManager( + scope = this, + bufferSize = 0, + upstream = upstreamFlow, + piggybackingDownstream = true, + keepUpstreamAlive = false, + onEach = { }, + ) + val firstChannel = Channel>(Channel.UNLIMITED) + val secondChannel = Channel>(Channel.UNLIMITED) + + channelManager.addDownstream(firstChannel) + advanceUntilIdle() + + // Removing the last downstream makes the actor suspend in doRemove on + // producer.cancelAndJoin(). Adding the next downstream right away enqueues its + // AddChannel message ahead of the producer's UpstreamFinished message, so the add is + // processed while the dead producer reference is still set. + channelManager.removeDownstream(firstChannel) + channelManager.addDownstream(secondChannel) + advanceUntilIdle() + + val dispatchedValue = secondChannel.tryReceive().getOrNull() + assertEquals(1, dispatchedValue?.value) + + channelManager.close() + } + private fun createChannels(count: Int): List>> { return (1..count).map { Channel(Channel.UNLIMITED) } }