Support for concurrency and timers in chat

Author: laevandus

Sources/StreamCore/Concurrency/Atomic.swift

@@ -57,6 +57,14 @@ public final class Atomic<T>: @unchecked Sendable {
         queue.sync { _value = changes(_value) }
     }
 
+    public func mutate(_ changes: (_ value: inout T) -> Void) {
+        self.mutate { value in
+            var updated = value
+            changes(&updated)
+            return updated
+        }
+    }
+    
     /// Update the value safely.
     /// - Parameter changes: a block with changes. It should return a new value.
     public func callAsFunction(_ changes: (_ value: T) -> T) { mutate(changes) }

Sources/StreamCore/Utils/Timers.swift

@@ -5,7 +5,7 @@
 import Combine
 import Foundation
 
-public protocol StreamTimer {
+public protocol TimerScheduling {
     /// Schedules a new timer.
     ///
     /// - Parameters:
@@ -33,7 +33,7 @@ public protocol StreamTimer {
     static func currentTime() -> Date
 }
 
-extension StreamTimer {
+extension TimerScheduling {
     public static func currentTime() -> Date {
         Date()
     }
@@ -58,7 +58,7 @@ extension DispatchWorkItem: TimerControl {}
 extension DispatchWorkItem: @retroactive @unchecked Sendable {}
 
 /// Default real-world implementations of timers.
-public struct DefaultTimer: StreamTimer {
+public struct DefaultTimer: TimerScheduling {
     @discardableResult
     public static func schedule(
         timeInterval: TimeInterval,

Sources/StreamCore/WebSocket/Client/ConnectionRecoveryHandler.swift

@@ -24,7 +24,7 @@ public final class DefaultConnectionRecoveryHandler: ConnectionRecoveryHandler,
     private let eventNotificationCenter: EventNotificationCenter
     private let backgroundTaskScheduler: BackgroundTaskScheduler?
     private let internetConnection: InternetConnection
-    private let reconnectionTimerType: StreamTimer.Type
+    private let reconnectionTimerType: TimerScheduling.Type
     private let keepConnectionAliveInBackground: Bool
     private nonisolated(unsafe) var reconnectionStrategy: RetryStrategy
     private nonisolated(unsafe) var reconnectionTimer: TimerControl?
@@ -38,7 +38,7 @@ public final class DefaultConnectionRecoveryHandler: ConnectionRecoveryHandler,
         backgroundTaskScheduler: BackgroundTaskScheduler?,
         internetConnection: InternetConnection,
         reconnectionStrategy: RetryStrategy,
-        reconnectionTimerType: StreamTimer.Type,
+        reconnectionTimerType: TimerScheduling.Type,
         keepConnectionAliveInBackground: Bool
     ) {
         self.init(
@@ -63,7 +63,7 @@ public final class DefaultConnectionRecoveryHandler: ConnectionRecoveryHandler,
         backgroundTaskScheduler: BackgroundTaskScheduler?,
         internetConnection: InternetConnection,
         reconnectionStrategy: RetryStrategy,
-        reconnectionTimerType: StreamTimer.Type,
+        reconnectionTimerType: TimerScheduling.Type,
         keepConnectionAliveInBackground: Bool,
         reconnectionPolicies: [AutomaticReconnectionPolicy]
     ) {

Sources/StreamCore/WebSocket/Client/WebSocketClient.swift

@@ -158,7 +158,7 @@ public extension WebSocketClient {
     /// An object encapsulating all dependencies of `WebSocketClient`.
     struct Environment {
         typealias CreatePingController = (
-            _ timerType: StreamTimer.Type,
+            _ timerType: TimerScheduling.Type,
             _ timerQueue: DispatchQueue,
             _ webSocketClientType: WebSocketClientType
         ) -> WebSocketPingController
@@ -169,7 +169,7 @@ public extension WebSocketClient {
             _ callbackQueue: DispatchQueue
         ) -> WebSocketEngine
 
-        var timerType: StreamTimer.Type = DefaultTimer.self
+        var timerType: TimerScheduling.Type = DefaultTimer.self
 
         var createPingController: CreatePingController = WebSocketPingController.init
 
@@ -186,7 +186,7 @@ public extension WebSocketClient {
         public init() {}
 
         init(
-            timerType: StreamTimer.Type = DefaultTimer.self,
+            timerType: TimerScheduling.Type = DefaultTimer.self,
             createPingController: @escaping CreatePingController,
             createEngine: @escaping CreateEngine,
             eventBatcherBuilder: @escaping (

Sources/StreamCore/WebSocket/Client/WebSocketPingController.swift

@@ -26,7 +26,7 @@ class WebSocketPingController {
     /// The time interval for pong timeout.
     static let pongTimeoutTimeInterval: TimeInterval = 3
 
-    private let timerType: StreamTimer.Type
+    private let timerType: TimerScheduling.Type
     private let timerQueue: DispatchQueue
 
     /// The timer used for scheduling `ping` calls
@@ -51,7 +51,7 @@ class WebSocketPingController {
     ///   - timerType: a timer type.
     ///   - timerQueue: a timer dispatch queue.
     init(
-        timerType: StreamTimer.Type,
+        timerType: TimerScheduling.Type,
         timerQueue: DispatchQueue,
         webSocketClientType: WebSocketClientType
     ) {

Sources/StreamCore/WebSocket/Events/EventBatcher.swift

@@ -13,7 +13,7 @@ protocol EventBatcher: Sendable {
     var currentBatch: Batch { get }
 
     /// Creates new batch processor.
-    init(period: TimeInterval, timerType: StreamTimer.Type, handler: @escaping BatchHandler)
+    init(period: TimeInterval, timerType: TimerScheduling.Type, handler: @escaping BatchHandler)
 
     /// Adds the item to the current batch of events. If it's the first event also schedules batch processing
     /// that will happen when `period` has passed.
@@ -29,7 +29,7 @@ class Batcher<Item>: @unchecked Sendable {
     /// The batching period. If the item is added sonner then `period` has passed after the first item they will get into the same batch.
     private let period: TimeInterval
     /// The time used to create timers.
-    private let timerType: StreamTimer.Type
+    private let timerType: TimerScheduling.Type
     /// The timer that  calls `processor` when fired.
     private let batchProcessingTimer = AllocatedUnfairLock<TimerControl?>(nil)
     /// The closure which processes the batch.
@@ -41,7 +41,7 @@ class Batcher<Item>: @unchecked Sendable {
 
     init(
         period: TimeInterval,
-        timerType: StreamTimer.Type = DefaultTimer.self,
+        timerType: TimerScheduling.Type = DefaultTimer.self,
         handler: @Sendable @escaping (_ batch: [Item], _ completion: @Sendable @escaping () -> Void) -> Void
     ) {
         self.period = max(period, 0)

Tests/StreamCoreTests/Mocks/EventBatcher_Mock.swift

@@ -10,7 +10,7 @@ final class EventBatcher_Mock: Batcher<Event>, @unchecked Sendable {
 
     override init(
         period: TimeInterval = 0,
-        timerType: StreamTimer.Type = DefaultTimer.self,
+        timerType: TimerScheduling.Type = DefaultTimer.self,
         handler: @escaping @Sendable (_ batch: [Event], _ completion: @escaping @Sendable () -> Void) -> Void
     ) {
         self.handler = handler

Tests/StreamCoreTests/TestUtils/VirtualTime/VirtualTimer.swift

@@ -5,7 +5,7 @@
 import Foundation
 @testable import StreamCore
 
-struct VirtualTimeTimer: StreamTimer {
+struct VirtualTimeTimer: TimerScheduling {
     nonisolated(unsafe) static var time: VirtualTime!
 
     static func invalidate() {