From dcedd62d00d3c17745162f0b6f728aad2be7c235 Mon Sep 17 00:00:00 2001 From: Ilias Pavlidakis Date: Fri, 26 Jun 2026 12:06:29 +0300 Subject: [PATCH] [Enhancement]Extend callJoinInterceptor hook flow --- .../Components/CallJoinInterceptor.swift | 2 +- .../DemoCallJoinInterceptor.swift | 94 +++++++++++++++++++ .../Stages/Call+JoinedStage.swift | 12 +++ .../Stages/Call+JoiningStage.swift | 36 +++++++ .../Controllers/CallController.swift | 11 +++ .../Utils/CallJoinIntercepting.swift | 42 +++++++++ .../CallStateMachine_JoinedStageTests.swift | 36 +++++++ .../CallStateMachine_JoiningStageTests.swift | 39 ++++++++ .../Mock/MockCallController.swift | 9 ++ 9 files changed, 280 insertions(+), 1 deletion(-) diff --git a/DemoApp/Sources/Components/Debug/Items/FeatureFlags/Components/CallJoinInterceptor.swift b/DemoApp/Sources/Components/Debug/Items/FeatureFlags/Components/CallJoinInterceptor.swift index c10e41d72..72d533e0c 100644 --- a/DemoApp/Sources/Components/Debug/Items/FeatureFlags/Components/CallJoinInterceptor.swift +++ b/DemoApp/Sources/Components/Debug/Items/FeatureFlags/Components/CallJoinInterceptor.swift @@ -33,7 +33,7 @@ extension AppEnvironment { } /// Default to enabled in debug builds so the pipeline is exercised during development. - nonisolated(unsafe) static var callJoinInterceptor: CallJoinInterceptor = .none + nonisolated(unsafe) static var callJoinInterceptor: CallJoinInterceptor = .synchronised } extension DebugMenu { diff --git a/DemoApp/Sources/Components/DemoCallJoinInterceptor/DemoCallJoinInterceptor.swift b/DemoApp/Sources/Components/DemoCallJoinInterceptor/DemoCallJoinInterceptor.swift index 1f5d03e3d..6f6336297 100644 --- a/DemoApp/Sources/Components/DemoCallJoinInterceptor/DemoCallJoinInterceptor.swift +++ b/DemoApp/Sources/Components/DemoCallJoinInterceptor/DemoCallJoinInterceptor.swift @@ -21,6 +21,17 @@ final class DemoCallJoinInterceptor: CallJoinIntercepting { private let hasOtherReadyParticipants: CurrentValueSubject = .init(false) + /// Creates the interceptor and starts tracking the active ringing call. + /// + /// The readiness handshake is only meaningful for ringing (1:1/group) + /// calls, so the interceptor observes `streamVideo.state.ringingCall` from + /// construction. Each time the ringing call changes it (re)wires the + /// custom-event observation in `didUpdate(ringingCall:)`. The current user + /// id is cached here because it's used both to emit the local readiness + /// marker and to ignore the marker echoed back from ourselves. + /// + /// - Parameter customEventKey: The custom-event key used to exchange the + /// readiness marker between participants. init( customEventKey: String = "participant.ready" ) { @@ -42,6 +53,18 @@ final class DemoCallJoinInterceptor: CallJoinIntercepting { // MARK: - CallJoinIntercepting + /// Mutes incoming media while the call is still connecting. + /// + /// Called by the SDK when the WebRTC layer enters its joining stage. The + /// demo silences remote audio so the local user doesn't hear other + /// participants until the readiness handshake completes and the call is + /// fully joined, avoiding leaking audio during the "preparing" window. + /// + /// - Parameter call: The call being prepared for joining. + func callWillJoin(_ call: Call) async { + await disableMediaTracksAndCallSettings(call) + } + /// Sends the local readiness marker and waits until another participant /// sends the same signal. /// @@ -67,8 +90,31 @@ final class DemoCallJoinInterceptor: CallJoinIntercepting { .nextValue() } + /// Restores incoming media once the call is fully joined. + /// + /// Counterpart to ``callWillJoin(_:)``. Called by the SDK after the call + /// has transitioned into the joined state, it re-enables the remote audio + /// that was muted during preparation and stops the deactivation observer so + /// the user can finally hear the other participants. + /// + /// - Parameter call: The call that just became active. + func callDidJoin(_ call: Call) async { + await restoreMediaTracksAndCallSettings(call) + } + // MARK: - Private Helpers + /// Rewires the readiness observation whenever the ringing call changes. + /// + /// Any previous observation is cancelled first so a stale subscription + /// never leaks across calls. When a new ringing call is present, it + /// subscribes to that call's custom events, keeps only the readiness marker + /// emitted by *other* participants (filtering out our own echo), and flips + /// `hasOtherReadyParticipants` to `true` — which is what + /// ``callReadyToJoin(_:)`` is waiting on. When the ringing call becomes + /// `nil` it simply leaves the observation cancelled. + /// + /// - Parameter ringingCall: The currently ringing call, or `nil` if none. private func didUpdate(ringingCall: Call?) { cancelCustomEventObservation() @@ -92,6 +138,12 @@ final class DemoCallJoinInterceptor: CallJoinIntercepting { log.debug("Call presence events observation has started.") } + /// Tears down the readiness observation and resets its state. + /// + /// Used when the ringing call changes or disappears. It cancels the custom + /// event subscription and resets `hasOtherReadyParticipants` to `false` so + /// a later join doesn't inherit a stale "ready" signal from a previous + /// call. The early return keeps the call idempotent when nothing is active. private func cancelCustomEventObservation() { guard customEventCancellable != nil else { return @@ -101,4 +153,46 @@ final class DemoCallJoinInterceptor: CallJoinIntercepting { hasOtherReadyParticipants.send(false) log.debug("Call presence events observation was cancelled.") } + + /// Continuously mutes remote participants' audio while preparing. + /// + /// A simple one-shot mute isn't enough: participants (and their audio + /// tracks) can arrive after the join begins. This subscribes to the + /// participants list and disables the audio track of every participant + /// other than the local user as they appear, so newly published remote + /// audio stays muted too. The subscription is keyed in the disposable bag + /// so ``restoreMediaTracksAndCallSettings(_:)`` can stop it once joined. + /// + /// - Parameter call: The call whose remote audio should be muted. + private func disableMediaTracksAndCallSettings(_ call: Call) async { + let currentUser = call.state.localParticipant + + call + .state + .$participants + .map { array in array.filter { $0.sessionId != currentUser?.sessionId } } + .map { $0.compactMap { $0.audioTrack } } + .filter { $0.isEmpty == false } + .sink { $0.forEach { $0.isEnabled = false } } + .store(in: disposableBag, key: "tracks-deactivation") + } + + /// Re-enables remote participants' audio after the call has joined. + /// + /// Reverses ``disableMediaTracksAndCallSettings(_:)``: it turns the audio + /// track of every remote participant back on for a one-time restore, then + /// removes the keyed subscription that was muting newly arriving tracks so + /// the demo stops interfering with normal call audio. + /// + /// - Parameter call: The call whose remote audio should be restored. + private func restoreMediaTracksAndCallSettings(_ call: Call) async { + let currentUser = call.state.localParticipant + call + .state + .participants + .filter { $0.sessionId != currentUser?.sessionId && $0.audioTrack != nil } + .forEach { $0.audioTrack?.isEnabled = true } + + disposableBag.remove("tracks-deactivation") + } } diff --git a/Sources/StreamVideo/CallStateMachine/Stages/Call+JoinedStage.swift b/Sources/StreamVideo/CallStateMachine/Stages/Call+JoinedStage.swift index d18c6f036..7facf195a 100644 --- a/Sources/StreamVideo/CallStateMachine/Stages/Call+JoinedStage.swift +++ b/Sources/StreamVideo/CallStateMachine/Stages/Call+JoinedStage.swift @@ -19,6 +19,7 @@ extension Call.StateMachine.Stage { JoinedStage( .init( call: context.call, + input: context.input, output: .joined(response) ) ) @@ -77,11 +78,22 @@ extension Call.StateMachine.Stage { /// /// The method subscribes to call settings and capability updates so the /// call controller and media managers stay aligned with live call state. + /// Once those subscriptions are in place it notifies the optional join + /// interceptor via ``CallJoinIntercepting/callDidJoin(_:)``, allowing it + /// to undo any temporary setup applied during the preparing stage. The + /// interceptor is recovered from the join input that travelled through + /// the state machine context. private func execute() { + let joinInterceptor: CallJoinIntercepting? = { + guard case let .join(input) = context.input else { return nil } + return input.joinInterceptor + }() + Task(disposableBag: disposableBag) { [weak self] in guard let self, let call = context.call else { return } await subscribeToCallSettingsUpdates(on: call) await subscribeToOwnCapabilitiesChanges(on: call) + await joinInterceptor?.callDidJoin(call) } } diff --git a/Sources/StreamVideo/CallStateMachine/Stages/Call+JoiningStage.swift b/Sources/StreamVideo/CallStateMachine/Stages/Call+JoiningStage.swift index a837adb25..be653fd5a 100644 --- a/Sources/StreamVideo/CallStateMachine/Stages/Call+JoiningStage.swift +++ b/Sources/StreamVideo/CallStateMachine/Stages/Call+JoiningStage.swift @@ -141,6 +141,10 @@ extension Call.StateMachine.Stage { ) async throws -> JoinCallResponse { try Task.checkCancellation() + observeCallStage(call, joinCallInterceptor: input.joinInterceptor) + + try Task.checkCancellation() + let response = try await call.callController.joinCall( create: input.create, callSettings: input.callSettings, @@ -254,5 +258,37 @@ extension Call.StateMachine.Stage { throw CancellationError() } } + + /// Bridges the WebRTC join lifecycle to the interceptor's preparation + /// hook. + /// + /// When an interceptor is provided, this subscribes to the call + /// controller's `stagePublisher` and, every time the underlying WebRTC + /// state machine reports the `.joining` stage, invokes + /// ``CallJoinIntercepting/callWillJoin(_:)``. This gives integrators + /// a chance to run setup work (e.g. muting remote tracks) while the peer + /// connection is still being negotiated. The subscription is stored in + /// the stage-scoped `disposableBag`, so it is torn down automatically + /// when the stage transitions away. + /// + /// - Parameters: + /// - call: The call being joined. + /// - joinCallInterceptor: The optional interceptor to notify. When + /// `nil`, no observation is set up. + private func observeCallStage( + _ call: Call, + joinCallInterceptor: CallJoinIntercepting? + ) { + guard let joinCallInterceptor else { + return + } + + call + .callController + .stagePublisher + .filter { $0 == .joining } + .sinkTask(storeIn: disposableBag) { _ in await joinCallInterceptor.callWillJoin(call) } + .store(in: disposableBag) + } } } diff --git a/Sources/StreamVideo/Controllers/CallController.swift b/Sources/StreamVideo/Controllers/CallController.swift index d7e5d0cf1..77d980ef0 100644 --- a/Sources/StreamVideo/Controllers/CallController.swift +++ b/Sources/StreamVideo/Controllers/CallController.swift @@ -61,6 +61,17 @@ class CallController: @unchecked Sendable { } } + /// Publishes the identifier of the WebRTC coordinator's current state + /// machine stage. + /// + /// A new value is emitted whenever the join/connection lifecycle advances + /// (for example into `.joining` or `.joined`). The call's join state + /// machine observes this to drive the optional ``CallJoinIntercepting`` + /// preparation hook without reaching into WebRTC internals directly. + var stagePublisher: AnyPublisher { + webRTCCoordinator.stateMachine.publisher.map(\.id).eraseToAnyPublisher() + } + private let user: User private let callId: String private let callType: String diff --git a/Sources/StreamVideo/Utils/CallJoinIntercepting.swift b/Sources/StreamVideo/Utils/CallJoinIntercepting.swift index 652326f9c..b1870e036 100644 --- a/Sources/StreamVideo/Utils/CallJoinIntercepting.swift +++ b/Sources/StreamVideo/Utils/CallJoinIntercepting.swift @@ -20,6 +20,21 @@ public final class CallJoinInterceptionError: ClientError, @unchecked Sendable { /// active and completes the join flow. public protocol CallJoinIntercepting: Sendable { + /// Notifies the interceptor that the SDK has started preparing the call for + /// a join. + /// + /// This fires as soon as the underlying WebRTC state machine enters its + /// `.joining` stage, before the peer connection has finished negotiating. + /// Use it to apply setup that should be in place while the call is + /// connecting — for example muting remote media or tweaking local call + /// settings — so the user is never exposed to a half-connected experience. + /// + /// The hook is best-effort: the SDK does not wait for it before continuing + /// to negotiate the call. + /// + /// - Parameter call: The call that is being prepared for joining. + func callWillJoin(_ call: Call) async + /// Performs application-specific work before the SDK completes a call join. /// /// Return normally to let the call continue into the joined state. Throw an @@ -29,6 +44,33 @@ public protocol CallJoinIntercepting: Sendable { /// - Parameter call: The call whose join response has already been applied /// to local state and is about to become active. func callReadyToJoin(_ call: Call) async throws + + /// Notifies the interceptor that the call has fully transitioned into the + /// joined state. + /// + /// This is the counterpart to ``callWillJoin(_:)`` and is the right place + /// to undo any temporary setup applied while the call was connecting — for + /// example restoring remote media that was muted during preparation. + /// + /// Like ``callWillJoin(_:)`` it is best-effort and does not block the + /// call lifecycle. + /// + /// - Parameter call: The call that just became active. + func callDidJoin(_ call: Call) async +} + +/// Default no-op implementations that make the will-join and did-join hooks +/// optional. +/// +/// Only ``CallJoinIntercepting/callReadyToJoin(_:)`` must be implemented. +/// Integrators that don't need to react to the will-join or did-join +/// transitions can omit these methods, and the additions stay source-compatible +/// with existing conformers. +extension CallJoinIntercepting { + + public func callWillJoin(_ call: Call) async { /* No-op by default. */ } + + public func callDidJoin(_ call: Call) async { /* No-op by default. */ } } extension WebRTCTrace { diff --git a/StreamVideoTests/CallStateMachine/CallStateMachine/Stages/CallStateMachine_JoinedStageTests.swift b/StreamVideoTests/CallStateMachine/CallStateMachine/Stages/CallStateMachine_JoinedStageTests.swift index 1effeb607..c7eb3057a 100644 --- a/StreamVideoTests/CallStateMachine/CallStateMachine/Stages/CallStateMachine_JoinedStageTests.swift +++ b/StreamVideoTests/CallStateMachine/CallStateMachine/Stages/CallStateMachine_JoinedStageTests.swift @@ -97,6 +97,42 @@ final class CallStateMachineStageJoinedStage_Tests: StreamVideoTestCase, @unchec .contains([.sendAudio]) ?? false } } + + // MARK: - JoinInterceptor + + func test_joiningTransition_joinInterceptorProvided_callDidJoinInvoked() async { + let joinInterceptor = JoinedTrackingInterceptor_Spy() + let context = Call.StateMachine.Stage.Context( + call: call, + input: .join( + .init( + create: true, + ring: false, + notify: false, + source: .inApp, + deliverySubject: .init(nil), + joinInterceptor: joinInterceptor + ) + ) + ) + subject = .joined(context, response: response) + + _ = subject.transition(from: .init(id: .joining, context: context)) + + await fulfillment { + joinInterceptor.recordedJoinedCIds == [self.call.cId] + } + } +} + +private final class JoinedTrackingInterceptor_Spy: CallJoinIntercepting, @unchecked Sendable { + @Atomic private(set) var recordedJoinedCIds: [String] = [] + + func callReadyToJoin(_ call: Call) async throws {} + + func callDidJoin(_ call: Call) async { + recordedJoinedCIds.append(call.cId) + } } extension Call.StateMachine.Stage.Context.Output { diff --git a/StreamVideoTests/CallStateMachine/CallStateMachine/Stages/CallStateMachine_JoiningStageTests.swift b/StreamVideoTests/CallStateMachine/CallStateMachine/Stages/CallStateMachine_JoiningStageTests.swift index ee42bdc46..fd1bb3a60 100644 --- a/StreamVideoTests/CallStateMachine/CallStateMachine/Stages/CallStateMachine_JoiningStageTests.swift +++ b/StreamVideoTests/CallStateMachine/CallStateMachine/Stages/CallStateMachine_JoiningStageTests.swift @@ -662,6 +662,40 @@ final class StreamCallStateMachineStageJoiningStage_Tests: StreamVideoTestCase, cancellable.cancel() } + func test_execute_whenWebRTCStageBecomesJoining_invokesInterceptorCallIsPreparing() async throws { + // The interceptor suspends in `callReadyToJoin`, keeping the stage in + // `.joining` so the stage-observation subscription stays alive while we + // assert that the preparing hook fired. + let joinInterceptor = NonCancellableJoinInterceptor_Spy() + callController.stageSubject.send(.joining) + callController.stub(for: .join, with: JoinCallResponse.dummy()) + let context = Call.StateMachine.Stage.Context( + call: call, + input: .join( + .init( + create: true, + callSettings: .init(audioOn: false), + options: .init(memberIds: [.unique]), + ring: true, + notify: false, + source: .inApp, + deliverySubject: .init(nil), + joinInterceptor: joinInterceptor + ) + ) + ) + subject.transition = { self.transitionedToStage = $0 } + subject.context = context + + _ = subject.transition(from: .idle(.init())) + + await fulfilmentInMainActor(timeout: defaultTimeout) { + joinInterceptor.recordedPreparingCIds == [self.call.cId] + } + + joinInterceptor.resume() + } + func test_execute_withRetries_whenJoinFailsAndThereAreAvailableRetries_transitionsToJoining() async throws { let context = Call.StateMachine.Stage.Context( call: call, @@ -836,12 +870,17 @@ private final class CallJoinInterceptor_Spy: CallJoinIntercepting, @unchecked Se private final class NonCancellableJoinInterceptor_Spy: CallJoinIntercepting, @unchecked Sendable { private let onEntered: @Sendable () -> Void @Atomic private(set) var recordedCallCIds: [String] = [] + @Atomic private(set) var recordedPreparingCIds: [String] = [] @Atomic private var continuation: CheckedContinuation? init(onEntered: @escaping @Sendable () -> Void = {}) { self.onEntered = onEntered } + func callWillJoin(_ call: Call) async { + recordedPreparingCIds.append(call.cId) + } + func callReadyToJoin(_ call: Call) async throws { recordedCallCIds.append(call.cId) onEntered() diff --git a/StreamVideoTests/Mock/MockCallController.swift b/StreamVideoTests/Mock/MockCallController.swift index ddb2f67e3..00bdea027 100644 --- a/StreamVideoTests/Mock/MockCallController.swift +++ b/StreamVideoTests/Mock/MockCallController.swift @@ -2,6 +2,7 @@ // Copyright © 2026 Stream.io Inc. All rights reserved. // +import Combine @testable import StreamVideo final class MockCallController: CallController, Mockable, @unchecked Sendable { @@ -90,6 +91,14 @@ final class MockCallController: CallController, Mockable, @unchecked Sendable { } } + /// Backs ``stagePublisher`` so tests can drive WebRTC stage transitions + /// without standing up a real coordinator. + let stageSubject = CurrentValueSubject(.idle) + + override var stagePublisher: AnyPublisher { + stageSubject.eraseToAnyPublisher() + } + var stubbedProperty: [String: Any] = [:] var stubbedFunction: [FunctionKey: Any] = [:] @Atomic var stubbedFunctionInput: [FunctionKey: [MockFunctionInputKey]] = FunctionKey