diff --git a/build.gradle.kts b/build.gradle.kts index 9268f87a2ec..ee3dbd49a48 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -49,6 +49,7 @@ streamProject { coverage { includedModules = setOf( "stream-video-android-core", + "stream-video-android-ui-core", "stream-video-android-ui-compose", ) sonarExclusions = listOf( diff --git a/stream-video-android-ui-core/build.gradle.kts b/stream-video-android-ui-core/build.gradle.kts index dc7d6a0cbae..6e8fc5e85f9 100644 --- a/stream-video-android-ui-core/build.gradle.kts +++ b/stream-video-android-ui-core/build.gradle.kts @@ -52,4 +52,10 @@ dependencies { implementation(libs.androidx.lifecycle.runtime) implementation(libs.stream.log) + + testImplementation(libs.junit) + testImplementation(libs.mockk) + testImplementation(libs.turbine) + testImplementation(libs.kotlinx.coroutines.test) + testImplementation(libs.kotlin.test.junit) } diff --git a/stream-video-android-ui-core/src/main/kotlin/io/getstream/video/android/ui/common/StreamCallActivity.kt b/stream-video-android-ui-core/src/main/kotlin/io/getstream/video/android/ui/common/StreamCallActivity.kt index 103b3c556f0..475d577591d 100644 --- a/stream-video-android-ui-core/src/main/kotlin/io/getstream/video/android/ui/common/StreamCallActivity.kt +++ b/stream-video-android-ui-core/src/main/kotlin/io/getstream/video/android/ui/common/StreamCallActivity.kt @@ -78,9 +78,9 @@ import io.getstream.video.android.ui.common.StreamCallActivity.Companion.callInt import io.getstream.video.android.ui.common.models.StreamCallActivityException import io.getstream.video.android.ui.common.permission.PermissionManager import io.getstream.video.android.ui.common.util.StreamCallActivityDelicateApi +import io.getstream.video.android.ui.common.util.lastParticipantSignal import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.FlowPreview import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.async @@ -88,7 +88,6 @@ import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.collectLatest -import kotlinx.coroutines.flow.debounce import kotlinx.coroutines.flow.first import kotlinx.coroutines.isActive import kotlinx.coroutines.launch @@ -1478,15 +1477,17 @@ public abstract class StreamCallActivity : ComponentActivity(), ActivityCallOper /** * Processes participant leave events for the given [call]. * - * Observes [cachedCall.state.participants] and triggers [onLastParticipant] if only - * one or fewer participants remain. Debouncing is applied to handle quick network - * disconnect/reconnect scenarios. + * Observes [cachedCall.state.participants] together with the connection state and triggers + * [onLastParticipant] if only one or fewer participants remain while the connection is + * [RealtimeConnection.Connected]. Debouncing is applied to handle quick network + * disconnect/reconnect scenarios, and the check requires a connected state because the + * roster is unreliable while a join or reconnect is running and leaving would cancel it. + * See [lastParticipantSignal]. * * @param call the active [Call] associated with the event. * @param event the [VideoEvent] that triggered this processing, typically a [ParticipantLeftEvent] * or [CallSessionParticipantLeftEvent]. */ - @OptIn(FlowPreview::class) private fun processParticipantLeftEvent(call: Call, event: VideoEvent) { /** * - participantCountJob will be null when activity is newly created @@ -1494,25 +1495,23 @@ public abstract class StreamCallActivity : ComponentActivity(), ActivityCallOper */ if (participantCountJob == null) { participantCountJob = lifecycleScope.launch(supervisorJob) { - cachedCall.state.participants - /** - * A debounce is applied here to handle quick disconnect/reconnect scenarios - * caused by unstable network conditions. Without the debounce, other devices - * may receive a [ParticipantLeftEvent] prematurely, which could trigger - * unintended reactions in the call flow. - */ - .debounce(getParticipantUpdateDebounce(call)) - .collect { - logger.d { "Participant left, remaining: ${it.size}" } + lastParticipantSignal( + participants = cachedCall.state.participants, + connection = cachedCall.state.connection, + debounceMs = getParticipantUpdateDebounce(call), + onEvaluated = { roster, connection -> + logger.d { + "Participant left, remaining: ${roster.size}, connection: $connection" + } lifecycleScope.launch(Dispatchers.Default) { - it.forEachIndexed { i, v -> + roster.forEachIndexed { i, v -> logger.d { "Participant [$i]=${v.name.value}" } } } - if (it.size <= 1) { - onLastParticipant(call) - } - } + }, + ).collect { + onLastParticipant(call) + } } } } diff --git a/stream-video-android-ui-core/src/main/kotlin/io/getstream/video/android/ui/common/util/LastParticipantSignal.kt b/stream-video-android-ui-core/src/main/kotlin/io/getstream/video/android/ui/common/util/LastParticipantSignal.kt new file mode 100644 index 00000000000..1d8cd3ff543 --- /dev/null +++ b/stream-video-android-ui-core/src/main/kotlin/io/getstream/video/android/ui/common/util/LastParticipantSignal.kt @@ -0,0 +1,92 @@ +/* + * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved. + * + * Licensed under the Stream License; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://github.com/GetStream/stream-video-android/blob/main/LICENSE + * + * 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 io.getstream.video.android.ui.common.util + +import io.getstream.video.android.core.ParticipantState +import io.getstream.video.android.core.RealtimeConnection +import kotlinx.coroutines.FlowPreview +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.combine +import kotlinx.coroutines.flow.debounce +import kotlinx.coroutines.flow.mapNotNull +import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.flow.runningFold + +/** + * Emits the participant roster whenever the local user has become the last participant in the + * call while the connection is [RealtimeConnection.Connected]. + * + * Roster changes are debounced by [debounceMs] to absorb quick disconnect/reconnect flaps. + * Emissions require a connected state because the roster is unreliable at any other point: + * during the initial join it is still being populated, and during a reconnect the rejoin + * removes the previous local participant record and remote participants of the failing SFU + * may not have rejoined yet. Acting on the roster then would leave the call and cancel the + * join or reconnect that is still running in the call scope. Terminal states need no signal + * from here: the reconnector leaves the call itself when retries are exhausted. The + * connection state is part of the combined stream, so the roster is re-evaluated once the + * connection settles and a genuine last-participant state still emits. + * + * The signal is a rising edge of the roster, not of the combined condition: becoming the last + * participant arms it, and it fires on the first connected evaluation after that. A connection + * transition with an unchanged roster therefore does not repeat the signal, while a remote + * participant joining and leaving again re-arms it and does signal a second time. + * + * @param participants the participant roster of the call. + * @param connection the realtime connection state of the call. + * @param debounceMs debounce applied to the combined stream before evaluation. + * @param onEvaluated invoked for every debounced evaluation, regardless of the outcome. + */ +@OptIn(FlowPreview::class) +internal fun lastParticipantSignal( + participants: Flow>, + connection: Flow, + debounceMs: Long, + onEvaluated: suspend (List, RealtimeConnection) -> Unit = { _, _ -> }, +): Flow> = + combine(participants, connection) { roster, connectionState -> roster to connectionState } + .debounce(debounceMs) + .onEach { (roster, connectionState) -> onEvaluated(roster, connectionState) } + .runningFold(LastParticipantState()) { previous, (roster, connectionState) -> + val isLast = roster.size <= 1 + // Arm on the rising edge of the roster and stay armed until a connected + // evaluation consumes it, so a signal found mid-reconnect is not lost. + val armed = when { + !isLast -> false + !previous.wasLast -> true + else -> previous.armed + } + val connected = connectionState is RealtimeConnection.Connected + LastParticipantState( + wasLast = isLast, + armed = armed && !connected, + signal = roster.takeIf { armed && connected }, + ) + } + .mapNotNull { it.signal } + +/** + * Fold state of [lastParticipantSignal]. + * + * @param wasLast whether the previous evaluation saw a last-participant roster. + * @param armed whether a last-participant roster is waiting for a connected evaluation. + * @param signal the roster to emit for this evaluation, or null when there is nothing to emit. + */ +private data class LastParticipantState( + val wasLast: Boolean = false, + val armed: Boolean = false, + val signal: List? = null, +) diff --git a/stream-video-android-ui-core/src/test/kotlin/io/getstream/video/android/ui/common/util/LastParticipantSignalTest.kt b/stream-video-android-ui-core/src/test/kotlin/io/getstream/video/android/ui/common/util/LastParticipantSignalTest.kt new file mode 100644 index 00000000000..012d428f645 --- /dev/null +++ b/stream-video-android-ui-core/src/test/kotlin/io/getstream/video/android/ui/common/util/LastParticipantSignalTest.kt @@ -0,0 +1,234 @@ +/* + * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved. + * + * Licensed under the Stream License; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://github.com/GetStream/stream-video-android/blob/main/LICENSE + * + * 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 io.getstream.video.android.ui.common.util + +import app.cash.turbine.test +import io.getstream.video.android.core.ParticipantState +import io.getstream.video.android.core.RealtimeConnection +import io.mockk.mockk +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.runTest +import org.junit.Test +import kotlin.test.assertEquals + +/** + * Regression tests for the [lastParticipantSignal] gating (AND-1455): with + * `leaveWhenLastInCall = true`, a leave triggered while an SFU join or reconnect is running + * cancels the work in the call scope and leaves the UI stuck on "Connecting...". The signal + * must only act while the connection is connected, re-evaluate the roster once the connection + * settles, and not repeat an unchanged roster across connection transitions. + */ +@OptIn(ExperimentalCoroutinesApi::class) +internal class LastParticipantSignalTest { + + private val debounceMs = 1_000L + private val local = mockk() + private val remote = mockk() + + @Test + fun `emits when the last participant remains while connected`() = runTest { + val participants = MutableStateFlow(listOf(local, remote)) + val connection = MutableStateFlow(RealtimeConnection.Connected) + + lastParticipantSignal(participants, connection, debounceMs).test { + participants.value = listOf(local) + + advanceTimeBy(debounceMs + 1) + assertEquals(1, awaitItem().size) + } + } + + @Test + fun `does not emit while more than one participant is in the call`() = runTest { + val participants = MutableStateFlow(listOf(local, remote)) + val connection = MutableStateFlow(RealtimeConnection.Connected) + + lastParticipantSignal(participants, connection, debounceMs).test { + advanceTimeBy(debounceMs + 1) + expectNoEvents() + } + } + + @Test + fun `does not emit while the connection is reconnecting`() = runTest { + val participants = MutableStateFlow(listOf(local, remote)) + val connection = MutableStateFlow(RealtimeConnection.Connected) + + lastParticipantSignal(participants, connection, debounceMs).test { + connection.value = RealtimeConnection.Reconnecting + participants.value = listOf(local) + + advanceTimeBy(debounceMs + 1) + expectNoEvents() + } + } + + @Test + fun `does not emit while the connection is migrating`() = runTest { + val participants = MutableStateFlow(listOf(local, remote)) + val connection = MutableStateFlow(RealtimeConnection.Connected) + + lastParticipantSignal(participants, connection, debounceMs).test { + connection.value = RealtimeConnection.Migrating + participants.value = listOf(local) + + advanceTimeBy(debounceMs + 1) + expectNoEvents() + } + } + + @Test + fun `does not emit while the join is in progress`() = runTest { + val participants = MutableStateFlow(listOf(local)) + val connection = MutableStateFlow(RealtimeConnection.PreJoin) + + lastParticipantSignal(participants, connection, debounceMs).test { + advanceTimeBy(debounceMs + 1) + expectNoEvents() + + connection.value = RealtimeConnection.InProgress + advanceTimeBy(debounceMs + 1) + expectNoEvents() + } + } + + @Test + fun `does not emit when the reconnect fails terminally`() = runTest { + val participants = MutableStateFlow(listOf(local)) + val connection = MutableStateFlow(RealtimeConnection.Reconnecting) + + lastParticipantSignal(participants, connection, debounceMs).test { + connection.value = RealtimeConnection.ReconnectingFailed + + advanceTimeBy(debounceMs + 1) + expectNoEvents() + } + } + + @Test + fun `does not emit the same roster again after a reconnect flap`() = runTest { + val roster = listOf(local) + val participants = MutableStateFlow(roster) + val connection = MutableStateFlow(RealtimeConnection.Connected) + + lastParticipantSignal(participants, connection, debounceMs).test { + advanceTimeBy(debounceMs + 1) + assertEquals(1, awaitItem().size) + + connection.value = RealtimeConnection.Reconnecting + advanceTimeBy(debounceMs + 1) + connection.value = RealtimeConnection.Connected + advanceTimeBy(debounceMs + 1) + expectNoEvents() + } + } + + @Test + fun `emits again when a remote participant joins and leaves again`() = runTest { + val participants = MutableStateFlow(listOf(local)) + val connection = MutableStateFlow(RealtimeConnection.Connected) + + lastParticipantSignal(participants, connection, debounceMs).test { + advanceTimeBy(debounceMs + 1) + assertEquals(1, awaitItem().size) + + participants.value = listOf(local, remote) + advanceTimeBy(debounceMs + 1) + expectNoEvents() + + participants.value = listOf(local) + advanceTimeBy(debounceMs + 1) + assertEquals(1, awaitItem().size) + } + } + + @Test + fun `emits after the reconnect settles with the local user still alone`() = runTest { + val participants = MutableStateFlow(listOf(local, remote)) + val connection = MutableStateFlow(RealtimeConnection.Connected) + + lastParticipantSignal(participants, connection, debounceMs).test { + connection.value = RealtimeConnection.Reconnecting + participants.value = listOf(local) + advanceTimeBy(debounceMs + 1) + expectNoEvents() + + connection.value = RealtimeConnection.Connected + advanceTimeBy(debounceMs + 1) + assertEquals(1, awaitItem().size) + } + } + + @Test + fun `does not emit after the reconnect settles with the roster restored`() = runTest { + val participants = MutableStateFlow(listOf(local, remote)) + val connection = MutableStateFlow(RealtimeConnection.Connected) + + lastParticipantSignal(participants, connection, debounceMs).test { + connection.value = RealtimeConnection.Reconnecting + participants.value = listOf(local) + advanceTimeBy(debounceMs + 1) + expectNoEvents() + + participants.value = listOf(local, remote) + connection.value = RealtimeConnection.Connected + advanceTimeBy(debounceMs + 1) + expectNoEvents() + } + } + + @Test + fun `debounce absorbs a quick roster flap`() = runTest { + val participants = MutableStateFlow(listOf(local, remote)) + val connection = MutableStateFlow(RealtimeConnection.Connected) + + lastParticipantSignal(participants, connection, debounceMs).test { + participants.value = listOf(local) + advanceTimeBy(debounceMs - 1) + participants.value = listOf(local, remote) + + advanceTimeBy(debounceMs + 1) + expectNoEvents() + } + } + + @Test + fun `invokes onEvaluated for suppressed evaluations`() = runTest { + val participants = MutableStateFlow(listOf(local)) + val connection = MutableStateFlow(RealtimeConnection.Reconnecting) + val evaluations = mutableListOf>() + + lastParticipantSignal( + participants = participants, + connection = connection, + debounceMs = debounceMs, + onEvaluated = { roster, connectionState -> + evaluations += roster.size to connectionState + }, + ).test { + advanceTimeBy(debounceMs + 1) + expectNoEvents() + } + + assertEquals( + listOf(1 to RealtimeConnection.Reconnecting as RealtimeConnection), + evaluations, + ) + } +}