[Enhancement]Implement new ICE restart Spec (#869)

Author: ipavlidakis

Sources/StreamVideo/WebRTC/v2/PeerConnection/Adapters/ICEConnectionStateAdapter.swift

@@ -0,0 +1,104 @@
+//
+// Copyright © 2025 Stream.io Inc. All rights reserved.
+//
+
+import Combine
+import Foundation
+import StreamWebRTC
+
+/// An adapter that observes ICE connection state changes from the WebRTC peer
+/// connection and schedules reconnection attempts when necessary.
+final class ICEConnectionStateAdapter {
+
+    /// The peer connection coordinator to observe for ICE state changes.
+    weak var peerConnectionCoordinator: RTCPeerConnectionCoordinator? {
+        didSet { didUpdate(peerConnectionCoordinator) }
+    }
+
+    /// A cancellable for the ICE state subscription.
+    private var peerConnectionICEStateCancellable: AnyCancellable?
+
+    /// A cancellable for the scheduled ICE restart timer.
+    private var scheduledRestartICECancellable: AnyCancellable?
+
+    /// The interval after which to restart ICE when disconnected.
+    private let scheduleICERestartInterval: TimeInterval
+
+    /// A serial queue for processing ICE state events.
+    private let processingQueue = DispatchQueue(
+        label: "io.getstream.ICEConnectionStateAdapter"
+    )
+
+    /// Creates a new ICEConnectionStateAdapter instance.
+    ///
+    /// - Parameter scheduleICERestartInterval: The time interval after a
+    ///   disconnection to schedule an ICE restart. Default is 3 seconds.
+    init(scheduleICERestartInterval: TimeInterval = 3) {
+        self.scheduleICERestartInterval = scheduleICERestartInterval
+    }
+
+    /// Responds to ICE connection state changes and performs the appropriate
+    /// action depending on the new state.
+    ///
+    /// - Parameter state: The current ICE connection state.
+    ///   - If the state is `.failed`, ICE will be restarted immediately.
+    ///   - If the state is `.disconnected`, ICE will be scheduled to restart
+    ///     after a short delay.
+    ///   - If the state is `.connected`, any pending restart will be cancelled.
+    private func didUpdate(_ state: RTCIceConnectionState) {
+        switch state {
+        case .failed:
+            scheduledRestartICECancellable?.cancel()
+            scheduledRestartICECancellable = nil
+            peerConnectionCoordinator?.restartICE()
+
+        case .disconnected:
+            scheduledRestartICECancellable = Foundation
+                .Timer
+                .publish(every: scheduleICERestartInterval, on: .main, in: .default)
+                .autoconnect()
+                .sink { [weak self] _ in self?.restartICE() }
+
+        case .connected:
+            scheduledRestartICECancellable?.cancel()
+            scheduledRestartICECancellable = nil
+
+        default:
+            break
+        }
+    }
+
+    /// Cancels any scheduled ICE restart and restarts ICE immediately.
+    private func restartICE() {
+        scheduledRestartICECancellable?.cancel()
+        scheduledRestartICECancellable = nil
+        peerConnectionCoordinator?.restartICE()
+    }
+
+    /// Subscribes to ICE connection state events from the coordinator and
+    /// cancels previous subscriptions when the coordinator changes.
+    ///
+    /// - Parameter peerConnectionCoordinator: The coordinator to observe.
+    private func didUpdate(
+        _ peerConnectionCoordinator: RTCPeerConnectionCoordinator?
+    ) {
+        guard let peerConnectionCoordinator else {
+            peerConnectionICEStateCancellable?.cancel()
+            peerConnectionICEStateCancellable = nil
+
+            scheduledRestartICECancellable?.cancel()
+            scheduledRestartICECancellable = nil
+
+            return
+        }
+
+        peerConnectionICEStateCancellable = peerConnectionCoordinator
+            .eventPublisher
+            .compactMap { $0 as? StreamRTCPeerConnection.ICEConnectionChangedEvent }
+            .map(\.state)
+            .receive(on: processingQueue)
+            .removeDuplicates()
+            .log(.debug) { "ICE connection state changed to: \($0)" }
+            .sink { [weak self] in self?.didUpdate($0) }
+    }
+}

Sources/StreamVideo/WebRTC/v2/PeerConnection/Protocols/StreamRTCPeerConnectionProtocol.swift

@@ -25,6 +25,18 @@ protocol StreamRTCPeerConnectionProtocol: AnyObject, Sendable {
     /// A publisher for RTCPeerConnectionEvents.
     var publisher: AnyPublisher<RTCPeerConnectionEvent, Never> { get }
 
+    /// The current ICE connection state of the peer connection.
+    ///
+    /// This property reflects the state of the ICE (Interactive Connectivity Establishment) agent,
+    /// indicating the progress and status of the connection between peers.
+    var iceConnectionState: RTCIceConnectionState { get }
+
+    /// The current signaling state of the peer connection.
+    ///
+    /// This property reflects the overall state of the peer connection, including the signaling
+    /// process and the establishment of media channels.
+    var connectionState: RTCPeerConnectionState { get }
+
     /// Sets the local description asynchronously.
     /// - Parameter sessionDescription: The RTCSessionDescription to set as the local description.
     /// - Throws: An error if setting the local description fails.

Sources/StreamVideo/WebRTC/v2/PeerConnection/RTCPeerConnectionCoordinator.swift

@@ -48,6 +48,7 @@ class RTCPeerConnectionCoordinator: @unchecked Sendable {
     private let mediaAdapter: MediaAdapter
     private let iceAdapter: ICEAdapter
     private let sfuAdapter: SFUAdapter
+    private let iceConnectionStateAdapter: ICEConnectionStateAdapter
 
     private var callSettings: CallSettings
 
@@ -73,12 +74,30 @@ class RTCPeerConnectionCoordinator: @unchecked Sendable {
     var trackPublisher: AnyPublisher<TrackEvent, Never> { mediaAdapter.trackPublisher }
     var disconnectedPublisher: AnyPublisher<Void, Never> {
         peerConnection
-            .publisher(eventType: StreamRTCPeerConnection.ICEConnectionChangedEvent.self)
+            .publisher(eventType: StreamRTCPeerConnection.DidChangeConnectionStateEvent.self)
             .filter { $0.state == .disconnected || $0.state == .failed }
             .map { _ in () }
             .eraseToAnyPublisher()
     }
 
+    /// A Boolean value indicating whether the peer connection is in a healthy state.
+    ///
+    /// The peer connection is considered healthy if its ICE connection state is not
+    /// `.failed` or `.closed`, and its overall connection state is not `.failed` or `.closed`.
+    /// This property provides a quick way to check if the connection is active and able to
+    /// send or receive data.
+    var isHealthy: Bool {
+        let invalidICEConnectionStates = Set([RTCIceConnectionState.failed, .closed])
+        let invalidConnectionStates = Set([RTCPeerConnectionState.failed, .closed])
+        guard
+            !invalidICEConnectionStates.contains(peerConnection.iceConnectionState),
+            !invalidConnectionStates.contains(peerConnection.connectionState)
+        else {
+            return false
+        }
+        return true
+    }
+
     /// Retrieves track information for a specified track type and collection type.
     ///
     /// - Parameters:
@@ -143,7 +162,14 @@ class RTCPeerConnectionCoordinator: @unchecked Sendable {
                 publishOptions: publishOptions,
                 videoCaptureSessionProvider: videoCaptureSessionProvider,
                 screenShareSessionProvider: screenShareSessionProvider
-            )
+            ),
+            iceAdapter: .init(
+                sessionID: sessionId,
+                peerType: peerType,
+                peerConnection: peerConnection,
+                sfuAdapter: sfuAdapter
+            ),
+            iceConnectionStateAdapter: .init()
         )
     }
 
@@ -156,7 +182,9 @@ class RTCPeerConnectionCoordinator: @unchecked Sendable {
         audioSettings: AudioSettings,
         publishOptions: PublishOptions,
         sfuAdapter: SFUAdapter,
-        mediaAdapter: MediaAdapter
+        mediaAdapter: MediaAdapter,
+        iceAdapter: ICEAdapter,
+        iceConnectionStateAdapter: ICEConnectionStateAdapter
     ) {
         self.sessionId = sessionId
         self.peerType = peerType
@@ -170,13 +198,11 @@ class RTCPeerConnectionCoordinator: @unchecked Sendable {
             ? .peerConnectionPublisher
             : .peerConnectionSubscriber
         self.mediaAdapter = mediaAdapter
+        self.iceAdapter = iceAdapter
+        self.iceConnectionStateAdapter = iceConnectionStateAdapter
 
-        iceAdapter = .init(
-            sessionID: sessionId,
-            peerType: peerType,
-            peerConnection: peerConnection,
-            sfuAdapter: sfuAdapter
-        )
+        // Warm up instances
+        iceConnectionStateAdapter.peerConnectionCoordinator = self
 
         // Start ICERestart events observation
         observeICERestartEvents()

Sources/StreamVideo/WebRTC/v2/PeerConnection/StreamRTCPeerConnection.swift

@@ -38,6 +38,10 @@ final class StreamRTCPeerConnection: StreamRTCPeerConnectionProtocol, @unchecked
         .receive(on: dispatchQueue)
         .eraseToAnyPublisher()
 
+    var iceConnectionState: RTCIceConnectionState { source.iceConnectionState }
+
+    var connectionState: RTCPeerConnectionState { source.connectionState }
+
     private let delegatePublisher = DelegatePublisher()
     private let source: RTCPeerConnection
 

Sources/StreamVideo/WebRTC/v2/StateMachine/Stages/WebRTCCoordinator+Disconnected.swift

@@ -143,14 +143,22 @@ extension WebRTCCoordinator.StateMachine.Stage {
         }
 
         /// Attempts to reconnect based on the current reconnection strategy.
-        private func reconnect() {
+        private func reconnect() async {
             do {
                 switch context.reconnectionStrategy {
-                case let .fast(disconnectedSince, deadline) where abs(disconnectedSince.timeIntervalSinceNow) <= deadline:
-                    try transition?(.fastReconnecting(context))
-                case .fast, .rejoin:
+                case let .fast(disconnectedSince, deadline):
+                    if await isFastReconnectPossible(disconnectedSince: disconnectedSince, deadline: deadline) {
+                        context.fastReconnectionAttempts += 1
+                        try transition?(.fastReconnecting(context))
+                    } else {
+                        context.fastReconnectionAttempts = 0
+                        try transition?(.rejoining(context))
+                    }
+                case .rejoin:
+                    context.fastReconnectionAttempts = 0
                     try transition?(.rejoining(context))
                 case .migrate:
+                    context.fastReconnectionAttempts = 0
                     try transition?(.migrating(context))
                 case .unknown:
                     if let error = context.flowError {
@@ -191,7 +199,7 @@ extension WebRTCCoordinator.StateMachine.Stage {
                         .trace(.init(status: $0))
 
                     if $0.isAvailable {
-                        self?.reconnect()
+                        await self?.reconnect()
                     }
                 }
         }
@@ -239,5 +247,40 @@ extension WebRTCCoordinator.StateMachine.Stage {
         private func didTimeInStageExpired() {
             transitionErrorOrLog(ClientError.NetworkNotAvailable())
         }
+
+        /// Checks if a fast reconnect is possible based on several conditions.
+        ///
+        /// A fast reconnect is a lightweight process to restore a connection without
+        /// going through the full rejoin flow. This method evaluates whether the current
+        /// state of the coordinator allows for such a reconnection.
+        ///
+        /// - Parameters:
+        ///   - disconnectedSince: The `Date` when the disconnection occurred.
+        ///   - deadline: The `TimeInterval` within which a fast reconnect must be
+        ///               initiated.
+        /// - Returns: `true` if a fast reconnect is possible, `false` otherwise.
+        private func isFastReconnectPossible(
+            disconnectedSince: Date,
+            deadline: TimeInterval
+        ) async -> Bool {
+            guard
+                // Ensure we haven't exceeded the maximum number of fast reconnection attempts.
+                context.fastReconnectionAttempts < context.fastReconnectionMaxAttempts,
+                // Check if the time since disconnection is within the allowed deadline.
+                abs(disconnectedSince.timeIntervalSinceNow) <= deadline,
+                // Verify that the WebRTC publisher is available and in a healthy state.
+                let publisher = await context.coordinator?.stateAdapter.publisher,
+                publisher.isHealthy,
+                // Verify that the WebRTC subscriber is available and in a healthy state.
+                let subscriber = await context.coordinator?.stateAdapter.subscriber,
+                subscriber.isHealthy
+            else {
+                // If any of the conditions are not met, fast reconnect is not possible.
+                return false
+            }
+
+            // All conditions are met, so a fast reconnect is possible.
+            return true
+        }
     }
 }