Handling many participant joins (#170)

GetStream · Oct 4, 2023 · 8f2ffb6 · 8f2ffb6
1 parent 932804f
commit 8f2ffb6
Show file tree

Hide file tree

Showing 5 changed files with 61 additions and 15 deletions.
diff --git a/Sources/StreamVideo/Utils/Sorting.swift b/Sources/StreamVideo/Utils/Sorting.swift
@@ -7,7 +7,7 @@ import Foundation
 public typealias StreamSortComparator<Value> = (Value, Value) -> ComparisonResult
 
 public let defaultComparators: [StreamSortComparator<CallParticipant>] = [
-    pinned, screensharing, dominantSpeaker, publishingVideo, publishingAudio, userId
+    pinned, screensharing, dominantSpeaker, publishingVideo, publishingAudio, joinedAt, userId
 ]
 
 public let livestreamComparators: [StreamSortComparator<CallParticipant>] = [
@@ -64,6 +64,10 @@ public var userId: StreamSortComparator<CallParticipant> = { (p1, p2) in
     p1.id >= p2.id ? .orderedDescending : .orderedAscending
 }
 
+public var joinedAt: StreamSortComparator<CallParticipant> = { (p1, p2) in
+    p1.joinedAt <= p2.joinedAt ? .orderedDescending : .orderedAscending
+}
+
 public extension Sequence {
     func sorted(using comparators: [StreamSortComparator<Element>], order: SortOrder = .descending) -> [Element] {
         sorted { valueA, valueB in

diff --git a/Sources/StreamVideo/WebRTC/Retries.swift b/Sources/StreamVideo/WebRTC/Retries.swift
@@ -29,7 +29,7 @@ func executeTask<Output>(
         if retries < retryPolicy.maxRetries && shouldRetryError(error) {
             let delay = UInt64(retryPolicy.delay(retries) * 1_000_000_000)
             try await Task.sleep(nanoseconds: delay)
-            if retryPolicy.runPrecondition() {
+            if await retryPolicy.runPrecondition() {
                 return try await executeTask(
                     retryPolicy: retryPolicy,
                     task: task,
@@ -49,7 +49,7 @@ func executeTask<Output>(
 struct RetryPolicy {
     let maxRetries: Int
     let delay: (Int) -> TimeInterval
-    var runPrecondition: () -> Bool = { true }
+    var runPrecondition: () async -> Bool = { true }
 }
 
 extension RetryPolicy {
@@ -63,7 +63,7 @@ extension RetryPolicy {
         )
     }
 
-    static func neverGonnaGiveYouUp(_ condition: @escaping () -> Bool) -> RetryPolicy {
+    static func neverGonnaGiveYouUp(_ condition: @escaping () async -> Bool) -> RetryPolicy {
         RetryPolicy(
             maxRetries: 30,
             delay: { delay(retries: $0) },

diff --git a/Sources/StreamVideo/WebRTC/WebRTCClient.swift b/Sources/StreamVideo/WebRTC/WebRTCClient.swift
@@ -15,11 +15,27 @@ class WebRTCClient: NSObject, @unchecked Sendable {
     }
 
     actor State: ObservableObject {
+        enum Constants {
+            static let lowParticipantDelay: UInt64 = 250_000_000
+            static let mediumParticipantDelay: UInt64 = 500_000_000
+            static let highParticipantDelay: UInt64 = 1_000_000_000
+        }
+        private var scheduledUpdate = false
         private var cancellables = Set<AnyCancellable>()
+        private(set) var lastUpdate: TimeInterval = Date().timeIntervalSince1970
         var connectionState = ConnectionState.disconnected(reason: nil)
         @Published var callParticipants = [String: CallParticipant]() {
             didSet {
-                continuation?.yield([true])
+                if !scheduledUpdate {
+                    scheduledUpdate = true
+                    Task {
+                        try? await Task.sleep(nanoseconds: participantUpdatesDelay)
+                        lastUpdate = Date().timeIntervalSince1970
+                        continuation?.yield([true])
+                        scheduledUpdate = false
+                    }
+                }
+
             }
         }
         var tracks = [String: RTCVideoTrack]()
@@ -95,6 +111,19 @@ class WebRTCClient: NSObject, @unchecked Sendable {
             connectionState = .disconnected(reason: .user)
             continuation?.finish()
         }
+
+        private var participantUpdatesDelay: UInt64 {
+            let count = callParticipants.count
+            if count < 16 {
+                return 0
+            } else if count < 50 {
+                return Constants.lowParticipantDelay
+            } else if count < 100 {
+                return Constants.mediumParticipantDelay
+            } else {
+                return Constants.highParticipantDelay
+            }
+        }
     }
 
     let state: State
@@ -912,9 +941,10 @@ class WebRTCClient: NSObject, @unchecked Sendable {
         let connectionState = await state.connectionState
         if connectionState == .connected && !tracks.isEmpty {
             request.tracks = tracks
-            let connectURL = signalChannel?.connectURL
+            let lastUpdate = await state.lastUpdate
             try await executeTask(retryPolicy: .neverGonnaGiveYouUp { [weak self] in
-                self?.sfuChanged(connectURL) == false
+                let currentUpdate = await self?.state.lastUpdate
+                return currentUpdate == lastUpdate
             }) {
                 _ = try await signalService.updateSubscriptions(
                     updateSubscriptionsRequest: request

diff --git a/Sources/StreamVideoSwiftUI/CallView/VideoRenderer.swift b/Sources/StreamVideoSwiftUI/CallView/VideoRenderer.swift
@@ -105,6 +105,10 @@ public class VideoRenderer: RTCMTLVideoView {
         self.track?.trackId
     }
 
+    private var viewSize: CGSize?
+
+    private lazy var scale: CGFloat = UIScreen.main.scale
+
     public func add(track: RTCVideoTrack) {
         queue.sync {
             if track.trackId == self.track?.trackId && track.readyState == .live {
@@ -120,6 +124,14 @@ public class VideoRenderer: RTCMTLVideoView {
         }
     }
 
+    public override func layoutSubviews() {
+        super.layoutSubviews()
+        self.viewSize = CGSize(
+            width: self.bounds.size.width * scale,
+            height: self.bounds.size.height * scale
+        )
+    }
+
     public override func renderFrame(_ frame: RTCVideoFrame?) {
         super.renderFrame(frame)
 
@@ -171,15 +183,11 @@ extension VideoRenderer {
         if let track = participant.track {
             log.debug("adding track to a view \(self)")
             self.add(track: track)
-            DispatchQueue.main.asyncAfter(deadline: .now() + 0.01) {
+            DispatchQueue.global(qos: .userInteractive).asyncAfter(deadline: .now() + 0.01) { [weak self] in
+                guard let self else { return }
                 let prev = participant.trackSize
-                let scale = UIScreen.main.scale
-                let newSize = CGSize(
-                    width: self.bounds.size.width * scale,
-                    height: self.bounds.size.height * scale
-                )
-                if prev != newSize {
-                    onTrackSizeUpdate(newSize, participant)
+                if let viewSize, prev != viewSize {
+                    onTrackSizeUpdate(viewSize, participant)
                 }
             }
         }

diff --git a/Sources/StreamVideoSwiftUI/CallViewModel.swift b/Sources/StreamVideoSwiftUI/CallViewModel.swift
@@ -586,6 +586,10 @@ open class CallViewModel: ObservableObject {
                         }
                     }
                 } else if let participantEvent = callEventsHandler.checkForParticipantEvents(from: event) {
+                    guard participants.count < 25 else {
+                        log.debug("Skipping participant events for big calls")
+                        return
+                    }
                     self.participantEvent = participantEvent
                     if participantEvent.action == .leave &&
                         callParticipants.count == 1