Fix thread safety for RTCPeerConnection access

Author: sampepose

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

@@ -28,7 +28,7 @@ final class StreamRTCPeerConnection: StreamRTCPeerConnectionProtocol, @unchecked
     var subject: PassthroughSubject<RTCPeerConnectionEvent, Never> { delegatePublisher.publisher }
 
     /// A dispatch queue for handling peer connection operations.
-    let dispatchQueue = DispatchQueue(label: "io.getstream.peerconnection")
+    let dispatchQueue = DispatchQueue(label: "io.getstream.peerconnection.dispatch")
 
     /// A publisher for RTCPeerConnectionEvents.
     lazy var publisher: AnyPublisher<RTCPeerConnectionEvent, Never> = delegatePublisher
@@ -38,6 +38,8 @@ final class StreamRTCPeerConnection: StreamRTCPeerConnectionProtocol, @unchecked
 
     private let delegatePublisher = DelegatePublisher()
     private let source: RTCPeerConnection
+    /// A dispatch queue for safely accessing `source`. RTCPeerConnection is not thread-safe.
+    private let connectionQueue = DispatchQueue(label: "io.getstream.peerconnection.connection")
 
     /// Initializes a new StreamRTCPeerConnection.
     ///
@@ -83,11 +85,14 @@ final class StreamRTCPeerConnection: StreamRTCPeerConnectionProtocol, @unchecked
                 return
             }
 
-            source.setLocalDescription(sessionDescription) { error in
-                if let error = error {
-                    continuation.resume(throwing: error)
-                } else {
-                    continuation.resume(returning: ())
+            connectionQueue.async { [weak self] in
+                guard let self else { return }
+                source.setLocalDescription(sessionDescription) { error in
+                    if let error = error {
+                        continuation.resume(throwing: error)
+                    } else {
+                        continuation.resume(returning: ())
+                    }
                 }
             }
         } as ()
@@ -108,11 +113,14 @@ final class StreamRTCPeerConnection: StreamRTCPeerConnectionProtocol, @unchecked
                 return
             }
 
-            source.setRemoteDescription(sessionDescription) { error in
-                if let error = error {
-                    continuation.resume(throwing: error)
-                } else {
-                    continuation.resume(returning: ())
+            connectionQueue.async { [weak self] in
+                guard let self else { return }
+                source.setRemoteDescription(sessionDescription) { error in
+                    if let error = error {
+                        continuation.resume(throwing: error)
+                    } else {
+                        continuation.resume(returning: ())
+                    }
                 }
             }
         } as ()
@@ -126,7 +134,24 @@ final class StreamRTCPeerConnection: StreamRTCPeerConnectionProtocol, @unchecked
     func offer(
         for constraints: RTCMediaConstraints
     ) async throws -> RTCSessionDescription {
-        try await source.offer(for: constraints)
+        try await withCheckedThrowingContinuation { [weak self] continuation in
+            guard let self else {
+                continuation.resume(throwing: ClientError.Unknown("RTCPeerConnection instance is unavailable."))
+                return
+            }
+            
+            connectionQueue.async { [weak self] in
+                guard let self else { return }
+                Task {
+                    do {
+                        let offer = try await self.source.offer(for: constraints)
+                        continuation.resume(returning: offer)
+                    } catch {
+                        continuation.resume(throwing: error)
+                    }
+                }
+            }
+        }
     }
 
     /// Creates an answer asynchronously.
@@ -137,15 +162,45 @@ final class StreamRTCPeerConnection: StreamRTCPeerConnectionProtocol, @unchecked
     func answer(
         for constraints: RTCMediaConstraints
     ) async throws -> RTCSessionDescription {
-        try await source.answer(for: constraints)
+        try await withCheckedThrowingContinuation { [weak self] continuation in
+            guard let self else {
+                continuation.resume(throwing: ClientError.Unknown("RTCPeerConnection instance is unavailable."))
+                return
+            }
+            
+            connectionQueue.async { [weak self] in
+                guard let self else { return }
+                Task {
+                    do {
+                        let offer = try await self.source.answer(for: constraints)
+                        continuation.resume(returning: offer)
+                    } catch {
+                        continuation.resume(throwing: error)
+                    }
+                }
+            }
+        }
     }
 
     /// Retrieves the statistics of the peer connection.
     ///
     /// - Returns: An RTCStatisticsReport containing the connection statistics.
     /// - Throws: An error if retrieving statistics fails.
     func statistics() async throws -> RTCStatisticsReport? {
-        await source.statistics()
+        try await withCheckedThrowingContinuation { [weak self] continuation in
+            guard let self else {
+                continuation.resume(throwing: ClientError.Unknown("RTCPeerConnection instance is unavailable."))
+                return
+            }
+            
+            connectionQueue.async { [weak self] in
+                guard let self else { return }
+                Task {
+                    let statistics = await self.source.statistics()
+                    continuation.resume(returning: statistics)
+                }
+            }
+        }
     }
 
     // MARK: - Forwarding API
@@ -161,8 +216,13 @@ final class StreamRTCPeerConnection: StreamRTCPeerConnectionProtocol, @unchecked
         with track: RTCMediaStreamTrack,
         init transceiverInit: RTCRtpTransceiverInit
     ) -> RTCRtpTransceiver? {
-        let result = source.addTransceiver(with: track, init: transceiverInit)
-        storeTransceiver(result, trackType: trackType)
+        var result: RTCRtpTransceiver?
+        connectionQueue.sync {
+            result = source.addTransceiver(with: track, init: transceiverInit)
+        }
+        if let result {
+            storeTransceiver(result, trackType: trackType)
+        }
         return result
     }
 
@@ -175,7 +235,32 @@ final class StreamRTCPeerConnection: StreamRTCPeerConnectionProtocol, @unchecked
     /// - Parameter candidate: The ICE candidate to add.
     /// - Throws: An error if adding the candidate fails.
     func add(_ candidate: RTCIceCandidate) async throws {
-        try await source.add(candidate)
+        try await withCheckedThrowingContinuation { [weak self] (continuation: CheckedContinuation<Void, Error>) in
+            guard let self else {
+                continuation.resume(
+                    throwing: ClientError.Unknown("RTCPeerConnection instance is unavailable.")
+                )
+                return
+            }
+            
+            connectionQueue.async { [weak self] in
+                guard let self else {
+                    continuation.resume(
+                        throwing: ClientError.Unknown("RTCPeerConnection instance is unavailable.")
+                    )
+                    return
+                }
+                
+                Task {
+                    do {
+                        try await self.source.add(candidate)
+                        continuation.resume(returning: ())
+                    } catch {
+                        continuation.resume(throwing: error)
+                    }
+                }
+            }
+        }
     }
 
     // MARK: - Publishing API
@@ -194,18 +279,25 @@ final class StreamRTCPeerConnection: StreamRTCPeerConnectionProtocol, @unchecked
 
     /// Restarts the ICE gathering process.
     func restartIce() {
-        source.restartIce()
+        connectionQueue.async { [weak self] in
+            guard let self else { return }
+            self.source.restartIce()
+        }
     }
 
     /// Closes the peer connection.
     func close() async {
-        Task { @MainActor in
-            /// It's very important to close any transceivers **before** we close the connection, to make
-            /// sure that access to `RTCVideoTrack` properties, will be handled correctly. Otherwise
-            /// if we try to access any property/method on a `RTCVideoTrack` instance whose
-            /// peerConnection has closed, we will get blocked on the Main Thread.
-            source.transceivers.forEach { $0.stopInternal() }
-            source.close()
+        await withCheckedContinuation { continuation in
+            connectionQueue.async { [weak self] in
+                guard let self else { return }
+                /// It's very important to close any transceivers **before** we close the connection, to make
+                /// sure that access to `RTCVideoTrack` properties, will be handled correctly. Otherwise
+                /// if we try to access any property/method on a `RTCVideoTrack` instance whose
+                /// peerConnection has closed, we will get blocked on the Main Thread.
+                self.source.transceivers.forEach { $0.stopInternal() }
+                self.source.close()
+                continuation.resume()
+            }
         }
     }