[Enhancement]Reduce the number of active timers (#854)

Author: ipavlidakis

.swiftlint.yml

@@ -3,15 +3,22 @@ included:
   - Sources
   - DemoApp
   - DemoAppUIKit
-  - StreamVideoTests
-  - StreamVideoUIKitTests
 excluded:
   - Tests/SwiftLintFrameworkTests/Resources
   - Sources/StreamVideo/Generated
   - Sources/StreamVideo/OpenApi
   - Sources/StreamVideo/protobuf
   - Sources/StreamVideoSwiftUI/Generated
 
+# Custom Rules
+custom_rules:
+  discourage_timer_publish:
+    name: "Discouraged Timer.publish usage"
+    regex: '\bTimer\s*(\.\s*)?publish\b'
+    message: "Avoid using Timer.publish. User the TimerStorage instead."
+    severity: error
+
 # Enabled/disabled rules
 only_rules:
-  - unhandled_throwing_task
+  - unhandled_throwing_task
+  - custom_rules

Sources/StreamVideo/CallKit/CallKitService.swift

@@ -13,6 +13,7 @@ open class CallKitService: NSObject, CXProviderDelegate, @unchecked Sendable {
 
     @Injected(\.callCache) private var callCache
     @Injected(\.uuidFactory) private var uuidFactory
+    @Injected(\.timers) private var timers
 
     /// Represents a call that is being managed by the service.
     final class CallEntry: Equatable, @unchecked Sendable {
@@ -599,20 +600,16 @@ open class CallKitService: NSObject, CXProviderDelegate, @unchecked Sendable {
     /// - Parameter callState: The state of the call.
     open func setUpRingingTimer(for callState: GetCallResponse) {
         let timeout = TimeInterval(callState.call.settings.ring.autoCancelTimeoutMs / 1000)
-        ringingTimerCancellable = Foundation.Timer.publish(
-            every: timeout,
-            on: .main,
-            in: .default
-        )
-        .autoconnect()
-        .sink { [weak self] _ in
-            log.debug(
-                "Detected ringing timeout, hanging up...",
-                subsystems: .callKit
-            )
-            self?.callEnded(callState.call.cid, ringingTimedOut: true)
-            self?.ringingTimerCancellable = nil
-        }
+        ringingTimerCancellable = timers
+            .timer(for: timeout)
+            .sink { [weak self] _ in
+                log.debug(
+                    "Detected ringing timeout, hanging up...",
+                    subsystems: .callKit
+                )
+                self?.callEnded(callState.call.cid, ringingTimedOut: true)
+                self?.ringingTimerCancellable = nil
+            }
     }
 
     /// A method that's being called every time the StreamVideo instance is getting updated.

Sources/StreamVideo/CallState.swift

@@ -33,7 +33,8 @@ public struct PermissionRequest: @unchecked Sendable, Identifiable {
 public class CallState: ObservableObject {
 
     @Injected(\.streamVideo) var streamVideo
-    
+    @Injected(\.timers) var timers
+
     /// The id of the current session.
     /// When a call is started, a unique session identifier is assigned to the user in the call.
     @Published public internal(set) var sessionId: String = ""
@@ -153,7 +154,7 @@ public class CallState: ObservableObject {
     }
 
     private var localCallSettingsUpdate = false
-    private var durationTimer: Foundation.Timer?
+    private var durationCancellable: AnyCancellable?
 
     /// We mark this one as `nonisolated` to allow us to initialise a state instance without isolation.
     /// That's a safe operation because `MainActor` is only required to ensure that all `@Published`
@@ -497,36 +498,21 @@ public class CallState: ObservableObject {
 
     private func setupDurationTimer() {
         resetTimer()
-        durationTimer = Foundation.Timer.scheduledTimer(withTimeInterval: 1.0, repeats: true, block: { [weak self] timer in
-            guard let self else {
-                timer.invalidate()
-                return
-            }
-            Task {
-                await MainActor.run {
-                    self.updateDuration()
+        durationCancellable = timers
+            .timer(for: 1.0)
+            .receive(on: DispatchQueue.main)
+            .compactMap { [weak self] _ in
+                if let startedAt = self?.startedAt {
+                    return Date().timeIntervalSince(startedAt)
+                } else {
+                    return 0
                 }
             }
-        })
+            .assign(to: \.duration, onWeak: self)
     }
 
     private func resetTimer() {
-        durationTimer?.invalidate()
-        durationTimer = nil
-    }
-    
-    @objc private func updateDuration() {
-        guard let startedAt else {
-            update(duration: 0)
-            return
-        }
-        let timeInterval = Date().timeIntervalSince(startedAt)
-        update(duration: timeInterval)
-    }
-    
-    private func update(duration: TimeInterval) {
-        if duration != self.duration {
-            self.duration = duration
-        }
+        durationCancellable?.cancel()
+        durationCancellable = nil
     }
 }

Sources/StreamVideo/StreamVideo.swift

@@ -15,6 +15,7 @@ public typealias UserTokenUpdater = @Sendable(UserToken) -> Void
 public class StreamVideo: ObservableObject, @unchecked Sendable {
 
     @Injected(\.callCache) private var callCache
+    @Injected(\.timers) private var timers
 
     public final class State: ObservableObject, @unchecked Sendable {
         @Published public internal(set) var connection: ConnectionStatus
@@ -491,23 +492,16 @@ public class StreamVideo: ObservableObject, @unchecked Sendable {
         } else {
             throw ClientError.Unknown()
         }
-        var connected = false
-        var timeout = false
-        let control = DefaultTimer.schedule(timeInterval: 30, queue: .sdk) {
-            timeout = true
-        }
+        
         log.debug("Listening for WS connection")
-        webSocketClient?.onConnected = {
-            control.cancel()
-            connected = true
-            log.debug("WS connected")
-        }
 
-        while (!connected && !timeout) {
-            try await Task.sleep(nanoseconds: 100_000)
-        }
-        
-        if timeout {
+        do {
+            log.debug("Listening for WS connection")
+            _ = try await timers
+                .timer(for: 0.1)
+                .filter { [weak webSocketClient] _ in webSocketClient?.connectionState.isConnected == true }
+                .nextValue(timeout: 30)
+        } catch {
             log.debug("Timeout while waiting for WS connection opening")
             throw ClientError.NetworkError()
         }
@@ -557,18 +551,16 @@ public class StreamVideo: ObservableObject, @unchecked Sendable {
         }
         log.debug("Waiting for connection id")
 
-        while (loadConnectionIdFromHealthcheck() == nil && !timeout) {
-            try? await Task.sleep(nanoseconds: 100_000)
-        }
-        
-        control.cancel()
-        
-        if let connectionId = loadConnectionIdFromHealthcheck() {
-            log.debug("Connection id available from the WS")
-            return connectionId
+        do {
+            return try await timers
+                .timer(for: 0.1)
+                .log(.debug) { _ in "Waiting for connection id" }
+                .compactMap { [weak self] _ in self?.loadConnectionIdFromHealthcheck() }
+                .nextValue(timeout: 5)
+        } catch {
+            log.warning("Unable to load connectionId.")
+            return ""
         }
-        
-        return ""
     }
 
     private func loadConnectionIdFromHealthcheck() -> String? {

Sources/StreamVideo/Utils/AudioSession/AudioRecorder/StreamCallAudioRecorder.swift

@@ -16,6 +16,7 @@ open class StreamCallAudioRecorder: @unchecked Sendable {
 
     @Injected(\.activeCallProvider) private var activeCallProvider
     @Injected(\.activeCallAudioSession) private var activeCallAudioSession
+    @Injected(\.timers) private var timers
 
     /// The builder used to create the AVAudioRecorder instance.
     let audioRecorderBuilder: AVAudioRecorderBuilder
@@ -123,10 +124,8 @@ open class StreamCallAudioRecorder: @unchecked Sendable {
 
             updateMetersTimerCancellable?.cancel()
             disposableBag.remove("update-meters")
-            updateMetersTimerCancellable = Foundation
-                .Timer
-                .publish(every: 0.1, on: .main, in: .default)
-                .autoconnect()
+            updateMetersTimerCancellable = timers
+                .timer(for: ScreenPropertiesAdapter.currentValue.refreshRate)
                 .sinkTask(storeIn: disposableBag, identifier: "update-meters") { [weak self, audioRecorder] _ in
                     audioRecorder.updateMeters()
                     self?._metersPublisher.send(audioRecorder.averagePower(forChannel: 0))

Sources/StreamVideo/Utils/OrderedCapacityQueue/OrderedCapacityQueue.swift

@@ -8,6 +8,8 @@ import Foundation
 /// A thread-safe queue that maintains a fixed capacity and removes elements after
 /// a specified time interval.
 final class OrderedCapacityQueue<Element> {
+    @Injected(\.timers) private var timers
+
     private let queue = UnfairQueue()
 
     /// The maximum number of elements the queue can hold.
@@ -48,10 +50,8 @@ final class OrderedCapacityQueue<Element> {
     init(capacity: Int, removalTime: TimeInterval) {
         self.capacity = capacity
         self.removalTime = removalTime
-        removalTimerCancellable = Foundation
-            .Timer
-            .publish(every: ScreenPropertiesAdapter.currentValue.refreshRate, on: .main, in: .default)
-            .autoconnect()
+        removalTimerCancellable = timers
+            .timer(for: ScreenPropertiesAdapter.currentValue.refreshRate)
             .sink { [weak self] _ in self?.removeItemsIfRequired() }
     }
 
@@ -86,14 +86,8 @@ final class OrderedCapacityQueue<Element> {
     ///   should be enabled.
     private func toggleRemovalObservation(_ isEnabled: Bool) {
         if isEnabled, removalTimerCancellable == nil {
-            removalTimerCancellable = Foundation
-                .Timer
-                .publish(
-                    every: ScreenPropertiesAdapter.currentValue.refreshRate,
-                    on: .main,
-                    in: .default
-                )
-                .autoconnect()
+            removalTimerCancellable = timers
+                .timer(for: ScreenPropertiesAdapter.currentValue.refreshRate)
                 .sink { [weak self] _ in self?.removeItemsIfRequired() }
         } else if !isEnabled, removalTimerCancellable != nil {
             removalTimerCancellable?.cancel()

Sources/StreamVideo/Utils/ScreenPropertiesAdapter/ScreenPropertiesAdapter.swift

@@ -7,11 +7,11 @@ import Foundation
 import UIKit
 #endif
 
-final class ScreenPropertiesAdapter: @unchecked Sendable {
+public final class ScreenPropertiesAdapter: @unchecked Sendable {
 
-    private(set) var preferredFramesPerSecond: Int = 0
-    private(set) var refreshRate: TimeInterval = 0
-    private(set) var scale: CGFloat = 0
+    public private(set) var preferredFramesPerSecond: Int = 0
+    public private(set) var refreshRate: TimeInterval = 0
+    public private(set) var scale: CGFloat = 0
 
     init() {
         Task { @MainActor in
@@ -29,11 +29,11 @@ final class ScreenPropertiesAdapter: @unchecked Sendable {
 }
 
 extension ScreenPropertiesAdapter: InjectionKey {
-    nonisolated(unsafe) static var currentValue: ScreenPropertiesAdapter = .init()
+    public nonisolated(unsafe) static var currentValue: ScreenPropertiesAdapter = .init()
 }
 
 extension InjectedValues {
-    var screenProperties: ScreenPropertiesAdapter {
+    public var screenProperties: ScreenPropertiesAdapter {
         set { Self[ScreenPropertiesAdapter.self] = newValue }
         get { Self[ScreenPropertiesAdapter.self] }
     }