Chat web socket client

Sources/StreamCore/Errors/ErrorPayload.swift

@@ -36,7 +36,7 @@ public struct ErrorPayload: LocalizedError, Codable, CustomDebugStringConvertibl
 
 extension ErrorPayload {
     /// Returns `true` if the code determines that the token is expired.
-    public var isExpiredTokenError: Bool {
+    public var isTokenExpiredError: Bool {
         code == StreamErrorCode.expiredToken
     }
 

Sources/StreamCore/Errors/Errors.swift

@@ -94,13 +94,13 @@ extension ClientError: Equatable {
 
 extension ClientError {
     /// Returns `true` the stream code determines that the token is expired.
-    public var isExpiredTokenError: Bool {
-        errorPayload?.isExpiredTokenError == true
+    public var isTokenExpiredError: Bool {
+        errorPayload?.isTokenExpiredError == true || apiError?.isTokenExpiredError == true
     }
 
     /// Returns `true` if underlaying error is `ErrorPayload` with code is inside invalid token codes range.
     public var isInvalidTokenError: Bool {
-        errorPayload?.isInvalidTokenError == true || apiError?.isTokenExpiredError == true
+        errorPayload?.isInvalidTokenError == true || apiError?.isInvalidTokenError == true
     }
 }
 
@@ -140,13 +140,13 @@ extension Error {
 
 extension Error {
     public var isTokenExpiredError: Bool {
-        if let error = self as? APIError, ClosedRange.tokenInvalidErrorCodes ~= error.code {
+        if let error = self as? APIError, error.isTokenExpiredError {
             return true
         }
-        if let error = self as? ErrorPayload, error.isExpiredTokenError {
+        if let error = self as? ErrorPayload, error.isTokenExpiredError {
             return true
         }
-        if let error = self as? ClientError, error.isExpiredTokenError {
+        if let error = self as? ClientError, error.isTokenExpiredError {
             return true
         }
         return false
@@ -173,6 +173,23 @@ extension ClosedRange where Bound == Int {
     public static let clientErrorCodes: Self = 400...499
 }
 
+extension APIError {
+    /// Returns `true` if the code determines that the token is expired.
+    public var isTokenExpiredError: Bool {
+        code == StreamErrorCode.expiredToken
+    }
+
+    /// Returns `true` if code is within invalid token codes range.
+    public var isInvalidTokenError: Bool {
+        ClosedRange.tokenInvalidErrorCodes ~= code || code == StreamErrorCode.accessKeyInvalid
+    }
+
+    /// Returns `true` if status code is within client error codes range.
+    public var isClientError: Bool {
+        ClosedRange.clientErrorCodes ~= statusCode
+    }
+}
+
 struct APIErrorContainer: Codable {
     let error: APIError
 }

Sources/StreamCore/Utils/InternetConnection.swift

@@ -8,10 +8,10 @@ import Network
 
 extension Notification.Name {
     /// Posted when any the Internet connection update is detected (including quality updates).
-    static let internetConnectionStatusDidChange = Self("io.getstream.StreamChat.internetConnectionStatus")
+    static let internetConnectionStatusDidChange = Self("io.getstream.core.internetConnectionStatus")
 
     /// Posted only when the Internet connection availability is changed (excluding quality updates).
-    static let internetConnectionAvailabilityDidChange = Self("io.getstream.StreamChat.internetConnectionAvailability")
+    static let internetConnectionAvailabilityDidChange = Self("io.getstream.core.internetConnectionAvailability")
 }
 
 extension Notification {
@@ -26,9 +26,9 @@ extension Notification {
 ///
 /// Basically, it's a wrapper over legacy monitor based on `Reachability` (iOS 11 only)
 /// and default monitor based on `Network`.`NWPathMonitor` (iOS 12+).
-public final class InternetConnection: @unchecked Sendable {
+open class InternetConnection: @unchecked Sendable {
     /// The current Internet connection status.
-    @Published private(set) var status: InternetConnectionStatus {
+    @Published public private(set) var status: InternetConnectionStatus {
         didSet {
             guard oldValue != status else { return }
 
@@ -93,7 +93,7 @@ public protocol InternetConnectionDelegate: AnyObject {
 }
 
 /// A protocol for Internet connection monitors.
-public protocol InternetConnectionMonitor: AnyObject {
+public protocol InternetConnectionMonitor: AnyObject, Sendable {
     /// A delegate for receiving Internet connection events.
     var delegate: InternetConnectionDelegate? { get set }
 

Sources/StreamCore/WebSocket/Client/BackgroundTaskScheduler.swift

@@ -9,16 +9,12 @@ public protocol BackgroundTaskScheduler: Sendable {
     /// It's your responsibility to finish previously running task.
     ///
     /// Returns: `false` if system forbid background task, `true` otherwise
-    @MainActor
-    func beginTask(expirationHandler: (@Sendable () -> Void)?) -> Bool
-    @MainActor
+    func beginTask(expirationHandler: (@MainActor () -> Void)?) -> Bool
     func endTask()
-    @MainActor
     func startListeningForAppStateUpdates(
         onEnteringBackground: @escaping () -> Void,
         onEnteringForeground: @escaping () -> Void
     )
-    @MainActor
     func stopListeningForAppStateUpdates()
 
     var isAppActive: Bool { get }
@@ -28,37 +24,48 @@ public protocol BackgroundTaskScheduler: Sendable {
 import UIKit
 
 public class IOSBackgroundTaskScheduler: BackgroundTaskScheduler, @unchecked Sendable {
-    private let applicationStateAdapter = StreamAppStateAdapter()
-
-    private lazy var app: UIApplication? = // We can't use `UIApplication.shared` directly because there's no way to convince the compiler
+    private lazy var app: UIApplication? = {
+        // We can't use `UIApplication.shared` directly because there's no way to convince the compiler
         // this code is accessible only for non-extension executables.
         UIApplication.value(forKeyPath: "sharedApplication") as? UIApplication
+    }()
 
     /// The identifier of the currently running background task. `nil` if no background task is running.
     private var activeBackgroundTask: UIBackgroundTaskIdentifier?
+    private let queue = DispatchQueue(label: "io.getstream.IOSBackgroundTaskScheduler", target: .global())
 
-    public var isAppActive: Bool { applicationStateAdapter.state == .foreground }
-
     public init() {}
+
+    public var isAppActive: Bool {
+        StreamConcurrency.onMain {
+            self.app?.applicationState == .active
+        }
+    }
 
-    @MainActor
-    public func beginTask(expirationHandler: (@Sendable () -> Void)?) -> Bool {
-        activeBackgroundTask = app?.beginBackgroundTask { [weak self] in
-            self?._endTask()
-            expirationHandler?()
+    public func beginTask(expirationHandler: (@MainActor () -> Void)?) -> Bool {
+        // Only a single task is allowed at the same time
+        endTask()
+
+        guard let app else { return false }
+        let identifier = app.beginBackgroundTask { [weak self] in
+            self?.endTask()
+            StreamConcurrency.onMain {
+                expirationHandler?()
+            }
+        }
+        queue.sync {
+            self.activeBackgroundTask = identifier
         }
-        return activeBackgroundTask != .invalid
+        return identifier != .invalid
     }
 
-    @MainActor
     public func endTask() {
-        _endTask()
-    }
-
-    private func _endTask() {
-        if let activeTask = activeBackgroundTask {
-            app?.endBackgroundTask(activeTask)
-            activeBackgroundTask = nil
+        guard let app else { return }
+        queue.sync {
+            if let identifier = self.activeBackgroundTask {
+                self.activeBackgroundTask = nil
+                app.endBackgroundTask(identifier)
+            }
         }
     }
 
@@ -69,8 +76,10 @@ public class IOSBackgroundTaskScheduler: BackgroundTaskScheduler, @unchecked Sen
         onEnteringBackground: @escaping () -> Void,
         onEnteringForeground: @escaping () -> Void
     ) {
-        self.onEnteringForeground = onEnteringForeground
-        self.onEnteringBackground = onEnteringBackground
+        queue.sync {
+            self.onEnteringForeground = onEnteringForeground
+            self.onEnteringBackground = onEnteringBackground
+        }
 
         NotificationCenter.default.addObserver(
             self,
@@ -88,8 +97,10 @@ public class IOSBackgroundTaskScheduler: BackgroundTaskScheduler, @unchecked Sen
     }
 
     public func stopListeningForAppStateUpdates() {
-        onEnteringForeground = {}
-        onEnteringBackground = {}
+        queue.sync {
+            self.onEnteringForeground = {}
+            self.onEnteringBackground = {}
+        }
 
         NotificationCenter.default.removeObserver(
             self,
@@ -105,19 +116,17 @@ public class IOSBackgroundTaskScheduler: BackgroundTaskScheduler, @unchecked Sen
     }
 
     @objc private func handleAppDidEnterBackground() {
-        onEnteringBackground()
+        let callback = queue.sync { onEnteringBackground }
+        callback()
     }
 
     @objc private func handleAppDidBecomeActive() {
-        onEnteringForeground()
+        let callback = queue.sync { onEnteringForeground }
+        callback()
     }
 
     deinit {
-        Task { @MainActor [activeBackgroundTask, app] in
-            if let activeTask = activeBackgroundTask {
-                app?.endBackgroundTask(activeTask)
-            }
-        }
+        endTask()
     }
 }
 

Sources/StreamCore/WebSocket/Client/ConnectionRecoveryHandler.swift

@@ -6,7 +6,10 @@ import CoreData
 import Foundation
 
 /// The type that keeps track of active chat components and asks them to reconnect when it's needed
-public protocol ConnectionRecoveryHandler: ConnectionStateDelegate, Sendable {}
+public protocol ConnectionRecoveryHandler: ConnectionStateDelegate, Sendable {
+    func start()
+    func stop()
+}
 
 /// The type is designed to obtain missing events that happened in watched channels while user
 /// was not connected to the web-socket.
@@ -76,39 +79,42 @@ public final class DefaultConnectionRecoveryHandler: ConnectionRecoveryHandler,
         self.keepConnectionAliveInBackground = keepConnectionAliveInBackground
         self.reconnectionPolicies = reconnectionPolicies
 
-        subscribeOnNotifications()
+        start()
     }
 
-    deinit {
+    public func start() {
+        subscribeOnNotifications()
+    }
+
+    public func stop() {
         unsubscribeFromNotifications()
         cancelReconnectionTimer()
     }
+
+    deinit {
+        stop()
+    }
 }
 
 // MARK: - Subscriptions
 
 private extension DefaultConnectionRecoveryHandler {
     func subscribeOnNotifications() {
-        Task { @MainActor in
-            backgroundTaskScheduler?.startListeningForAppStateUpdates(
-                onEnteringBackground: { [weak self] in self?.appDidEnterBackground() },
-                onEnteringForeground: { [weak self] in self?.appDidBecomeActive() }
-            )
-
-            internetConnection.notificationCenter.addObserver(
-                self,
-                selector: #selector(internetConnectionAvailabilityDidChange(_:)),
-                name: .internetConnectionAvailabilityDidChange,
-                object: nil
-            )
-        }
+        backgroundTaskScheduler?.startListeningForAppStateUpdates(
+            onEnteringBackground: { [weak self] in self?.appDidEnterBackground() },
+            onEnteringForeground: { [weak self] in self?.appDidBecomeActive() }
+        )
+
+        internetConnection.notificationCenter.addObserver(
+            self,
+            selector: #selector(internetConnectionAvailabilityDidChange(_:)),
+            name: .internetConnectionAvailabilityDidChange,
+            object: nil
+        )
     }
 
     func unsubscribeFromNotifications() {
-        Task { @MainActor [backgroundTaskScheduler] in
-            backgroundTaskScheduler?.stopListeningForAppStateUpdates()
-        }
-
+        backgroundTaskScheduler?.stopListeningForAppStateUpdates()
         internetConnection.notificationCenter.removeObserver(
             self,
             name: .internetConnectionStatusDidChange,
@@ -121,13 +127,11 @@ private extension DefaultConnectionRecoveryHandler {
 
 extension DefaultConnectionRecoveryHandler {
     private func appDidBecomeActive() {
-        Task { @MainActor in
-            log.debug("App -> ✅", subsystems: .webSocket)
-
-            backgroundTaskScheduler?.endTask()
-
-            reconnectIfNeeded()
-        }
+        log.debug("App -> ✅", subsystems: .webSocket)
+
+        backgroundTaskScheduler?.endTask()
+
+        reconnectIfNeeded()
     }
 
     private func appDidEnterBackground() {
@@ -145,20 +149,18 @@ extension DefaultConnectionRecoveryHandler {
         }
 
         guard let scheduler = backgroundTaskScheduler else { return }
-
-        Task { @MainActor in
-            let succeed = scheduler.beginTask { [weak self] in
-                log.debug("Background task -> ❌", subsystems: .webSocket)
-
-                self?.disconnectIfNeeded()
-            }
-
-            if succeed {
-                log.debug("Background task -> ✅", subsystems: .webSocket)
-            } else {
-                // Can't initiate a background task, close the connection
-                disconnectIfNeeded()
-            }
+
+        let succeed = scheduler.beginTask { [weak self] in
+            log.debug("Background task -> ❌", subsystems: .webSocket)
+
+            self?.disconnectIfNeeded()
+        }
+
+        if succeed {
+            log.debug("Background task -> ✅", subsystems: .webSocket)
+        } else {
+            // Can't initiate a background task, close the connection
+            disconnectIfNeeded()
         }
     }
 
@@ -205,7 +207,7 @@ private extension DefaultConnectionRecoveryHandler {
         let state = webSocketClient.connectionState
 
         switch state {
-        case .connecting, .authenticating, .connected, .disconnecting:
+        case .connecting, .authenticating, .connected:
             log.debug("Will disconnect automatically from \(state) state", subsystems: .webSocket)
 
             return true