diff --git a/demo-app/src/main/kotlin/io/getstream/video/android/CallActivity.kt b/demo-app/src/main/kotlin/io/getstream/video/android/CallActivity.kt index bf4df4beff9..1481bacfc40 100644 --- a/demo-app/src/main/kotlin/io/getstream/video/android/CallActivity.kt +++ b/demo-app/src/main/kotlin/io/getstream/video/android/CallActivity.kt @@ -49,6 +49,7 @@ import io.getstream.video.android.util.StreamVideoInitHelper import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job +import kotlinx.coroutines.cancel import kotlinx.coroutines.flow.collectLatest import kotlinx.coroutines.flow.filterNotNull import kotlinx.coroutines.flow.flatMapLatest @@ -68,7 +69,9 @@ class CallActivity : ComposeStreamCallActivity() { var observeCallReadyToJoinJob: Job? = null var observeRingingJob: Job? = null private val previousRingingStates = ConcurrentHashMap.newKeySet() - override val callJoinInterceptor = DemoCallJoinInterceptor(previousRingingStates) + private val coroutineScope = CoroutineScope(Dispatchers.Default) + override val callJoinInterceptor = + DemoCallJoinInterceptor(previousRingingStates, coroutineScope) /** * This code is required to pass the UI-tests (as it hardcodes the configuration) @@ -207,6 +210,7 @@ class CallActivity : ComposeStreamCallActivity() { super.finish() observeCallReadyToJoinJob?.cancel() observeRingingJob?.cancel() + coroutineScope.cancel() previousRingingStates.clear() } } diff --git a/demo-app/src/main/kotlin/io/getstream/video/android/DemoCallJoinInterceptor.kt b/demo-app/src/main/kotlin/io/getstream/video/android/DemoCallJoinInterceptor.kt index 3c1ab350ffa..bcf2c73757c 100644 --- a/demo-app/src/main/kotlin/io/getstream/video/android/DemoCallJoinInterceptor.kt +++ b/demo-app/src/main/kotlin/io/getstream/video/android/DemoCallJoinInterceptor.kt @@ -20,9 +20,17 @@ import io.getstream.log.taggedLogger import io.getstream.video.android.CallActivity.Companion.USE_CALL_JOIN_INTERCEPTOR import io.getstream.video.android.core.Call import io.getstream.video.android.core.CallJoinInterceptor +import io.getstream.video.android.core.RealtimeConnection import io.getstream.video.android.core.RingingState +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.flow.asFlow import kotlinx.coroutines.flow.filter +import kotlinx.coroutines.flow.filterNotNull import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.flatMapLatest +import kotlinx.coroutines.flow.flatMapMerge +import kotlinx.coroutines.launch /** * Do the following changes before testing this flow @@ -31,6 +39,7 @@ import kotlinx.coroutines.flow.first */ class DemoCallJoinInterceptor( private val previousRingingStates: Set, + private val coroutineScope: CoroutineScope, ) : CallJoinInterceptor { private val logger by taggedLogger("DemoCallJoinInterceptor") @@ -38,6 +47,36 @@ class DemoCallJoinInterceptor( const val CALLER_READY_TO_JOIN_EVENT_TYPE = "caller_ready_join" const val CALLEE_READY_TO_JOIN_EVENT_TYPE = "callee_ready_join" } + private var observeJob: Job? = null + override suspend fun callWillJoin(call: Call) { + if (USE_CALL_JOIN_INTERCEPTOR) { + observeJob?.cancel() + observeJob = coroutineScope.launch { + val muteJob = launch { + call.state.participants + .flatMapLatest { participants -> + participants + .filter { !it.isLocal } + .asFlow() + .flatMapMerge { participant -> + participant.audioTrack.filterNotNull() + } + } + .collect { track -> + track.audio.setEnabled(false) + } + } + // Stop muting once the call reaches a terminal state, even if callDidJoin + // never runs (e.g. the interceptor aborts the join or it's left externally). + call.state.connection.first { + it is RealtimeConnection.Disconnected || + it is RealtimeConnection.Failed || + it is RealtimeConnection.ReconnectingFailed + } + muteJob.cancel() + } + } + } override suspend fun callReadyToJoin(call: Call) { if (USE_CALL_JOIN_INTERCEPTOR) { @@ -81,4 +120,16 @@ class DemoCallJoinInterceptor( } } } + + override suspend fun callDidJoin(call: Call) { + if (USE_CALL_JOIN_INTERCEPTOR) { + observeJob?.cancel() + observeJob = null + call.state.participants.value.filter { !it.isLocal } + .forEach { + val audioTrack = it.audioTrack.value + audioTrack?.audio?.setEnabled(true) + } + } + } } diff --git a/stream-video-android-core/api/stream-video-android-core.api b/stream-video-android-core/api/stream-video-android-core.api index 73760ca76ac..49fe341b99a 100644 --- a/stream-video-android-core/api/stream-video-android-core.api +++ b/stream-video-android-core/api/stream-video-android-core.api @@ -8989,7 +8989,11 @@ public final class io/getstream/video/android/core/CallJoinInterceptionException } public abstract interface class io/getstream/video/android/core/CallJoinInterceptor { + public fun callDidJoin (Lio/getstream/video/android/core/Call;Lkotlin/coroutines/Continuation;)Ljava/lang/Object; + public static synthetic fun callDidJoin$suspendImpl (Lio/getstream/video/android/core/CallJoinInterceptor;Lio/getstream/video/android/core/Call;Lkotlin/coroutines/Continuation;)Ljava/lang/Object; public abstract fun callReadyToJoin (Lio/getstream/video/android/core/Call;Lkotlin/coroutines/Continuation;)Ljava/lang/Object; + public fun callWillJoin (Lio/getstream/video/android/core/Call;Lkotlin/coroutines/Continuation;)Ljava/lang/Object; + public static synthetic fun callWillJoin$suspendImpl (Lio/getstream/video/android/core/CallJoinInterceptor;Lio/getstream/video/android/core/Call;Lkotlin/coroutines/Continuation;)Ljava/lang/Object; } public final class io/getstream/video/android/core/CallKt { diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/ActiveStateGate.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/ActiveStateGate.kt index 403ad040e27..2f621c03dae 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/ActiveStateGate.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/ActiveStateGate.kt @@ -83,7 +83,10 @@ internal class ActiveStateGate( val shouldProceed = invokeInterceptor(call, interceptor) if (!isActive) return@launch - if (shouldProceed) onReady() + if (shouldProceed) { + onReady() + invokeCallDidJoin(call, interceptor) + } cancelInterceptorJob() }, ) @@ -107,12 +110,25 @@ internal class ActiveStateGate( val shouldProceed = invokeInterceptor(call, interceptor) if (!isActive) return@launch - if (shouldProceed) onReady() + if (shouldProceed) { + onReady() + invokeCallDidJoin(call, interceptor) + } cleanup() }, ) } + private suspend fun invokeCallDidJoin(call: Call, interceptor: CallJoinInterceptor?) { + try { + interceptor?.callDidJoin(call) + } catch (ex: CancellationException) { + throw ex + } catch (ex: Exception) { + logger.e(ex) { "[callDidJoin] interceptor threw, proceeding" } + } + } + private suspend fun awaitPeerConnection(call: Call) { val start = System.currentTimeMillis() val result = diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt index 6fead716778..5fca37651ea 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt @@ -103,6 +103,7 @@ import io.getstream.video.android.core.utils.safeCallWithDefault import io.getstream.video.android.core.utils.toQueriedMembers import io.getstream.video.android.model.User import io.getstream.webrtc.android.ui.VideoTextureViewRenderer +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob @@ -625,6 +626,15 @@ public class Call( "is joined. MediaManager will not be initialised with server settings." } } + + try { + callJoinInterceptor?.callWillJoin(this) + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + logger.e(e) { "[callWillJoin] interceptor threw, proceeding" } + } + return result } if (result is Failure) { diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallJoinInterceptor.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallJoinInterceptor.kt index 3b873923309..7a8f5c86524 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallJoinInterceptor.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallJoinInterceptor.kt @@ -19,14 +19,24 @@ package io.getstream.video.android.core import kotlin.jvm.Throws /** - * Controls when a ringing call transitions to [RingingState.Active]. + * Controls when a ringing call transitions to [RingingState.Active], and provides + * best-effort hooks around the join lifecycle. * * Implement this to insert custom logic (e.g. waiting for user confirmation) between * the publisher peer connection becoming ready and the call going active. * Has no effect on non-ringing joins (livestream, direct join). + * + * Order: [callWillJoin] → [callReadyToJoin] → [callDidJoin]. */ public interface CallJoinInterceptor { + /** + * Called right after the [io.getstream.video.android.core.call.RtcSession] is created, + * before the call goes [RingingState.Active]. Exceptions are logged and swallowed — + * this hook is best-effort and never fails the join. + */ + public suspend fun callWillJoin(call: Call) {} + /** * Called when the SDK is ready to transition to [RingingState.Active]. * Suspend here to delay the transition; return to allow it to proceed. @@ -34,10 +44,16 @@ public interface CallJoinInterceptor { * Throw [CallJoinInterceptionException] to abort the join — the SDK will leave * the call cleanly * - * The SDK enforces a 5-second maximum — the transition proceeds automatically on timeout. + * The SDK enforces a 5-second maximum [INTERCEPTOR_TIMEOUT_MS] — the transition proceeds automatically on timeout. */ @Throws(CallJoinInterceptionException::class) public suspend fun callReadyToJoin(call: Call) + + /** + * Called once the call has transitioned to [RingingState.Active] (after [callReadyToJoin]). + * Exceptions are logged and swallowed — this hook is best-effort and never fails the join. + */ + public suspend fun callDidJoin(call: Call) {} } /**