[WIP] Extend CallJoinInterceptor with will-join and did-join hooks

demo-app/src/main/kotlin/io/getstream/video/android/DemoCallJoinInterceptor.kt

@@ -19,10 +19,18 @@ package io.getstream.video.android
 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.RingingState
+import io.getstream.video.android.core.call.interceptor.CallJoinLifecycleInterceptor
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+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,13 +39,31 @@ import kotlinx.coroutines.flow.first
  */
 class DemoCallJoinInterceptor(
     private val previousRingingStates: Set<RingingState>,
-) : CallJoinInterceptor {
+) : CallJoinLifecycleInterceptor {
     private val logger by taggedLogger("DemoCallJoinInterceptor")
 
     companion object {
         const val CALLER_READY_TO_JOIN_EVENT_TYPE = "caller_ready_join"
         const val CALLEE_READY_TO_JOIN_EVENT_TYPE = "callee_ready_join"
     }
+    var observeJob: Job? = null
+    val observerScope = CoroutineScope(Dispatchers.Default)
+    override suspend fun callWillJoin(call: Call) {
+        observeJob = observerScope.launch {
+            call.state.participants
+                .flatMapLatest { participants ->
+                    participants
+                        .filter { !it.isLocal }
+                        .asFlow()
+                        .flatMapMerge { participant ->
+                            participant.audioTrack.filterNotNull()
+                        }
+                }
+                .collect { track ->
+                    track.audio.setEnabled(false)
+                }
+        }
+    }
 
     override suspend fun callReadyToJoin(call: Call) {
         if (USE_CALL_JOIN_INTERCEPTOR) {
@@ -81,4 +107,13 @@ class DemoCallJoinInterceptor(
             }
         }
     }
+
+    override suspend fun callDidJoin(call: Call) {
+        observeJob?.cancel()
+        call.state.participants.value.filter { !it.isLocal }
+            .forEach {
+                val audioTrack = it.audioTrack.value
+                audioTrack?.audio?.setEnabled(true)
+            }
+    }
 }

stream-video-android-core/api/stream-video-android-core.api

@@ -10070,6 +10070,11 @@ public final class io/getstream/video/android/core/call/connection/StreamPeerCon
 	public final fun toggleAudioProcessing ()Z
 }
 
+public abstract interface class io/getstream/video/android/core/call/interceptor/CallJoinLifecycleInterceptor : io/getstream/video/android/core/CallJoinInterceptor {
+	public abstract fun callDidJoin (Lio/getstream/video/android/core/Call;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public abstract fun callWillJoin (Lio/getstream/video/android/core/Call;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+}
+
 public final class io/getstream/video/android/core/call/signal/socket/RTCEventMapper {
 	public static final field INSTANCE Lio/getstream/video/android/core/call/signal/socket/RTCEventMapper;
 	public final fun mapEvent (Lstream/video/sfu/event/SfuEvent;)Lio/getstream/video/android/core/events/SfuDataEvent;

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/ActiveStateGate.kt

@@ -18,6 +18,7 @@
 
 import androidx.lifecycle.AtomicReference
 import io.getstream.log.taggedLogger
+import io.getstream.video.android.core.call.interceptor.CallJoinLifecycleInterceptor
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
@@ -50,7 +51,7 @@
     internal fun awaitAndTransition(
         currentRingingState: RingingState,
         call: Call,
         interceptor: CallJoinInterceptor?,
         onReady: () -> Unit,
     ) {
         if (currentRingingState is RingingState.Active) {
@@ -70,7 +71,7 @@
         launchGate(call, interceptor, waitForPublisherConnection = isRingingCall, onReady)
     }
 
     private fun handleLegacyBehaviour(call: Call, onReady: () -> Unit, interceptor: CallJoinInterceptor?) {
         if (interceptorJob.get()?.isActive == true) return
 
         if (interceptor == null) {
@@ -83,15 +84,26 @@
                 val shouldProceed = invokeInterceptor(call, interceptor)
                 if (!isActive) return@launch
 
-                if (shouldProceed) onReady()
+                if (shouldProceed) {
+                    onReady()
+                    try {
+                        if (interceptor is CallJoinLifecycleInterceptor) {
+                            interceptor.callDidJoin(call)
+                        }
+                    } catch (ex: CancellationException) {
+                        throw ex
+                    } catch (ex: Exception) {
+                        logger.e(ex) { "[callDidJoin] interceptor threw, proceeding" }
+                    }
+                }
                 cancelInterceptorJob()
             },
         )
     }
 
     private fun launchGate(
         call: Call,
         interceptor: CallJoinInterceptor?,
         waitForPublisherConnection: Boolean,
         onReady: () -> Unit,
     ) {
@@ -107,7 +119,18 @@
                 val shouldProceed = invokeInterceptor(call, interceptor)
                 if (!isActive) return@launch
 
-                if (shouldProceed) onReady()
+                if (shouldProceed) {
+                    onReady()
+                    try {
+                        if (interceptor is CallJoinLifecycleInterceptor) {
+                            interceptor.callDidJoin(call)
+                        }
+                    } catch (ex: CancellationException) {
+                        throw ex
+                    } catch (ex: Exception) {
+                        logger.e(ex) { "[callDidJoin] interceptor threw, proceeding" }
+                    }
+                }
                 cleanup()
             },
         )
@@ -133,7 +156,7 @@
 
     private suspend fun invokeInterceptor(
         call: Call,
         interceptor: CallJoinInterceptor?,
     ): Boolean {
         val startTime = System.currentTimeMillis()
         logger.d { "[invokeInterceptor] start at $startTime" }

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt

@@ -71,6 +71,7 @@
 import io.getstream.video.android.core.call.audio.InputAudioFilter
 import io.getstream.video.android.core.call.connection.StreamPeerConnectionFactory
 import io.getstream.video.android.core.call.connection.Subscriber
+import io.getstream.video.android.core.call.interceptor.CallJoinLifecycleInterceptor
 import io.getstream.video.android.core.call.scope.ScopeProvider
 import io.getstream.video.android.core.call.scope.ScopeProviderImpl
 import io.getstream.video.android.core.call.utils.SoundInputProcessor
@@ -112,6 +113,7 @@
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.filterNotNull
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.map
@@ -141,7 +143,7 @@
         "fastReconnectDeadlineSeconds. This constant will be removed in a future release.",
     level = DeprecationLevel.WARNING,
 )
 const val sfuReconnectTimeoutMillis = 30_000
 
 /**
  * Outcome of a single reconnect attempt. Each reconnect method returns one of
@@ -203,9 +205,9 @@
 
     internal val scope = CoroutineScope(clientImpl.scope.coroutineContext + supervisorJob)
 
     // Must be initialized before `state` — CallState → SortedParticipantsState
     // launches a coroutine that reads `call.events` (leaking-this race).
     val events = MutableSharedFlow<VideoEvent>(extraBufferCapacity = 150)
 
     /** The call state contains all state such as the participant list, reactions etc */
     val state = CallState(client, this, user, scope)
@@ -272,8 +274,8 @@
      */
     private var isDestroyed = false
 
     /** Session handles all real time communication for video and audio */
     internal val session: MutableStateFlow<RtcSession?> = MutableStateFlow(null)
 
     var sessionId = UUID.randomUUID().toString()
     internal val unifiedSessionId = UUID.randomUUID().toString()
@@ -559,13 +561,13 @@
         return response
     }
 
     suspend fun join(
         create: Boolean = false,
         createOptions: CreateCallOptions? = null,
         ring: Boolean = false,
         notify: Boolean = false,
         hintHighScaleLivestreamPublisher: Boolean? = null,
         callJoinInterceptor: CallJoinInterceptor? = null,
     ): Result<RtcSession> {
         callAnalytics.joinAnalytics.onJoinFunctionStart()
         callAnalytics.mediaPermissionObserver.mediaPermissionStatus()
@@ -624,6 +626,11 @@
                             "is joined. MediaManager will not be initialised with server settings."
                     }
                 }
+
+                if (callJoinInterceptor is CallJoinLifecycleInterceptor) {
+                    callJoinInterceptor.callWillJoin(this)
+                }
+
                 return result
             }
             if (result is Failure) {
@@ -658,7 +665,7 @@
         members: List<String>,
         createOptions: CreateCallOptions? = CreateCallOptions(members),
         video: Boolean = isVideoEnabled(),
         callJoinInterceptor: CallJoinInterceptor? = null,
     ): Result<RtcSession> {
         logger.d { "[joinAndRing] #ringing; #track; members: $members, video: $video" }
         state.toggleJoinAndRingProgress(true)
@@ -909,7 +916,7 @@
      * @param strategy the initial reconnection strategy requested by the caller.
      * @param reason a human-readable reason for logging / tracing.
      */
     internal suspend fun reconnect(
         strategy: WebsocketReconnectStrategy,
         reason: String,
     ) {
@@ -1914,7 +1921,7 @@
     }
 
     @VisibleForTesting
     internal suspend fun joinRequest(
         create: CreateCallOptions? = null,
         location: String,
         migratingFrom: String? = null,
@@ -1998,7 +2005,7 @@
     /**
      * Should outlive both the call scope and the service scope and needs to be executed in the client-level scope.
      * Because the call scope or service scope may be cancelled or finished while the network request is still in flight
      * TODO: Run this in clientImpl.scope internally
      */
     suspend fun reject(reason: RejectReason? = null): Result<RejectCallResponse> {
         logger.d { "[reject] #ringing; rejectReason: $reason, call_id:$id" }

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallJoinInterceptor.kt

@@ -25,7 +25,14 @@
  * the publisher peer connection becoming ready and the call going active.
  * Has no effect on non-ringing joins (livestream, direct join).
  */
+@Deprecated(
+    message = "Use CallJoinLifecycleInterceptor instead",
+    replaceWith = ReplaceWith(
+        "CallJoinLifecycleInterceptor",
+        "io.getstream.video.android.core.call.interceptor.CallJoinLifecycleInterceptor",
+    ),
+)
 public interface CallJoinInterceptor {
 
     /**
      * Called when the SDK is ready to transition to [RingingState.Active].
@@ -34,7 +41,7 @@
      * 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)

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/interceptor/CallJoinLifecycleInterceptor.kt

@@ -0,0 +1,39 @@
+/*
+ * 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.core.call.interceptor
+
+import io.getstream.video.android.core.Call
+import io.getstream.video.android.core.CallJoinInterceptor
+
+/**
+ * Extends [CallJoinInterceptor] with hooks around the join lifecycle.
+ *
+ * Order: [callWillJoin] → [callReadyToJoin] → [callDidJoin].
+ * Only affects ringing joins.
+ */
+@Suppress("DEPRECATION")
+public interface CallJoinLifecycleInterceptor : CallJoinInterceptor {
+
+    /** Called right after the [io.getstream.video.android.core.call.RtcSession] is created, before the call goes [io.getstream.video.android.core.RingingState.Active]. */
+    public suspend fun callWillJoin(call: Call)
+
+    /**
+     * Called once the call has transitioned to [io.getstream.video.android.core.RingingState.Active]
+     * (after [CallJoinInterceptor.callReadyToJoin]).
+     */
+    public suspend fun callDidJoin(call: Call)
+}