V5: Use JSON encoders and decoders from StreamCore (#3860)

* V5: Use web-socket client from StreamCore (#3861)

Author: laevandus

DemoApp/Screens/DemoReminderListVC.swift

@@ -478,7 +478,7 @@ extension DemoReminderListVC: MessageReminderListControllerDelegate, EventsContr
         updateRemindersData()
     }
 
-    func eventsController(_ controller: EventsController, didReceiveEvent event: any StreamChat.Event) {
+    func eventsController(_ controller: EventsController, didReceiveEvent event: any Event) {
         if event is MessageReminderDueEvent {
             updateReminderListsWithNewNowDate()
         }

DemoApp/Screens/Livestream/DemoLivestreamChatChannelVC.swift

@@ -393,7 +393,7 @@ class DemoLivestreamChatChannelVC: _ViewController,
         messageListVC.scrollToBottomButton.content = .init(messages: skippedMessagesAmount, mentions: 0)
     }
 
-    func eventsController(_ controller: EventsController, didReceiveEvent event: any StreamChat.Event) {
+    func eventsController(_ controller: EventsController, didReceiveEvent event: any Event) {
         if event is NewMessagePendingEvent {
             if livestreamChannelController.isPaused {
                 pauseBannerView.setState(.resuming)

Package.swift

@@ -29,7 +29,7 @@ let package = Package(
     ],
     dependencies: [
         .package(url: "https://github.com/apple/swift-docc-plugin", exact: "1.0.0"),
-        .package(url: "https://github.com/GetStream/stream-core-swift.git", branch: "chat-errors")
+        .package(url: "https://github.com/GetStream/stream-core-swift.git", exact: "0.5.0")
     ],
     targets: [
         .target(

Sources/StreamChat/APIClient/RequestDecoder.swift

@@ -46,9 +46,9 @@ struct DefaultRequestDecoder: RequestDecoder {
         log.debug("URL request response: \(httpResponse), data:\n\(data.debugPrettyPrintedJSON))", subsystems: .httpRequests)
 
         guard httpResponse.statusCode < 300 else {
-            let serverError: ErrorPayload
+            let serverError: APIError
             do {
-                serverError = try JSONDecoder.default.decode(ErrorPayload.self, from: data)
+                serverError = try JSONDecoder.default.decode(APIError.self, from: data)
             } catch {
                 log
                     .error(
@@ -58,7 +58,7 @@ struct DefaultRequestDecoder: RequestDecoder {
                 throw ClientError.Unknown("Unknown error. Server response: \(httpResponse).")
             }
 
-            if serverError.isExpiredTokenError {
+            if serverError.isTokenExpiredError {
                 log.info("Request failed because of an expired token.", subsystems: .httpRequests)
                 throw ClientError.ExpiredToken()
             }

Sources/StreamChat/ChatClient+Environment.swift

@@ -25,15 +25,18 @@ extension ChatClient {
 
         var webSocketClientBuilder: (@Sendable (
             _ sessionConfiguration: URLSessionConfiguration,
-            _ requestEncoder: RequestEncoder,
             _ eventDecoder: AnyEventDecoder,
-            _ notificationCenter: EventNotificationCenter
+            _ notificationCenter: PersistentEventNotificationCenter
         ) -> WebSocketClient)? = {
-            WebSocketClient(
+            let wsEnvironment = WebSocketClient.Environment(eventBatchingPeriod: 0.5)
+            return WebSocketClient(
                 sessionConfiguration: $0,
-                requestEncoder: $1,
-                eventDecoder: $2,
-                eventNotificationCenter: $3
+                eventDecoder: $1,
+                eventNotificationCenter: $2,
+                webSocketClientType: .coordinator,
+                environment: wsEnvironment,
+                connectRequest: nil,
+                healthCheckBeforeConnected: true
             )
         }
 
@@ -57,7 +60,7 @@ extension ChatClient {
 
         var eventDecoderBuilder: @Sendable () -> EventDecoder = { EventDecoder() }
 
-        var notificationCenterBuilder: @Sendable (_ database: DatabaseContainer, _ manualEventHandler: ManualEventHandler?) -> EventNotificationCenter = { EventNotificationCenter(database: $0, manualEventHandler: $1) }
+        var notificationCenterBuilder: @Sendable (_ database: DatabaseContainer, _ manualEventHandler: ManualEventHandler?) -> PersistentEventNotificationCenter = { PersistentEventNotificationCenter(database: $0, manualEventHandler: $1) }
 
         var internetConnection: @Sendable (_ center: NotificationCenter, _ monitor: InternetConnectionMonitor) -> InternetConnection = {
             InternetConnection(notificationCenter: $0, monitor: $1)
@@ -76,16 +79,18 @@ extension ChatClient {
         var connectionRepositoryBuilder: @Sendable (
             _ isClientInActiveMode: Bool,
             _ syncRepository: SyncRepository,
+            _ webSocketEncoder: RequestEncoder?,
             _ webSocketClient: WebSocketClient?,
             _ apiClient: APIClient,
             _ timerType: TimerScheduling.Type
         ) -> ConnectionRepository = {
             ConnectionRepository(
                 isClientInActiveMode: $0,
                 syncRepository: $1,
-                webSocketClient: $2,
-                apiClient: $3,
-                timerType: $4
+                webSocketEncoder: $2,
+                webSocketClient: $3,
+                apiClient: $4,
+                timerType: $5
             )
         }
 
@@ -110,20 +115,18 @@ extension ChatClient {
         var connectionRecoveryHandlerBuilder: @Sendable (
             _ webSocketClient: WebSocketClient,
             _ eventNotificationCenter: EventNotificationCenter,
-            _ syncRepository: SyncRepository,
             _ backgroundTaskScheduler: BackgroundTaskScheduler?,
             _ internetConnection: InternetConnection,
             _ keepConnectionAliveInBackground: Bool
         ) -> ConnectionRecoveryHandler = {
             DefaultConnectionRecoveryHandler(
                 webSocketClient: $0,
                 eventNotificationCenter: $1,
-                syncRepository: $2,
-                backgroundTaskScheduler: $3,
-                internetConnection: $4,
+                backgroundTaskScheduler: $2,
+                internetConnection: $3,
                 reconnectionStrategy: DefaultRetryStrategy(),
                 reconnectionTimerType: DefaultTimer.self,
-                keepConnectionAliveInBackground: $5
+                keepConnectionAliveInBackground: $4
             )
         }
 

Sources/StreamChat/ChatClient.swift

@@ -52,7 +52,7 @@ public class ChatClient: @unchecked Sendable {
     private(set) var connectionRecoveryHandler: ConnectionRecoveryHandler?
 
     /// The notification center used to send and receive notifications about incoming events.
-    private(set) var eventNotificationCenter: EventNotificationCenter
+    private(set) var eventNotificationCenter: PersistentEventNotificationCenter
 
     /// The registry that contains all the attachment payloads associated with their attachment types.
     /// For the meantime this is a static property to avoid breaking changes. On v5, this can be changed.
@@ -99,6 +99,7 @@ public class ChatClient: @unchecked Sendable {
 
     /// The `WebSocketClient` instance `Client` uses to communicate with Stream WS servers.
     let webSocketClient: WebSocketClient?
+    let webSocketEncoder: RequestEncoder?
 
     /// The `DatabaseContainer` instance `Client` uses to store and cache data.
     let databaseContainer: DatabaseContainer
@@ -184,13 +185,13 @@ public class ChatClient: @unchecked Sendable {
             channelListUpdater
         )
         let webSocketClient = factory.makeWebSocketClient(
-            requestEncoder: webSocketEncoder,
             urlSessionConfiguration: urlSessionConfiguration,
             eventNotificationCenter: eventNotificationCenter
         )
         let connectionRepository = environment.connectionRepositoryBuilder(
             config.isClientInActiveMode,
             syncRepository,
+            webSocketEncoder,
             webSocketClient,
             apiClient,
             environment.timerType
@@ -207,6 +208,7 @@ public class ChatClient: @unchecked Sendable {
         self.databaseContainer = databaseContainer
         self.apiClient = apiClient
         self.webSocketClient = webSocketClient
+        self.webSocketEncoder = webSocketEncoder
         self.eventNotificationCenter = eventNotificationCenter
         self.offlineRequestsRepository = offlineRequestsRepository
         self.connectionRepository = connectionRepository
@@ -268,7 +270,6 @@ public class ChatClient: @unchecked Sendable {
         connectionRecoveryHandler = environment.connectionRecoveryHandlerBuilder(
             webSocketClient,
             eventNotificationCenter,
-            syncRepository,
             environment.backgroundTaskSchedulerBuilder(),
             environment.internetConnection(eventNotificationCenter, environment.internetMonitor),
             config.staysConnectedInBackground
@@ -718,7 +719,7 @@ extension ChatClient: AuthenticationRepositoryDelegate {
 }
 
 extension ChatClient: ConnectionStateDelegate {
-    func webSocketClient(_ client: WebSocketClient, didUpdateConnectionState state: WebSocketConnectionState) {
+    public func webSocketClient(_ client: WebSocketClient, didUpdateConnectionState state: WebSocketConnectionState) {
         connectionRepository.handleConnectionUpdate(
             state: state,
             onExpiredToken: { [weak self] in

Sources/StreamChat/ChatClientFactory.swift

@@ -65,13 +65,11 @@ class ChatClientFactory {
     }
 
     func makeWebSocketClient(
-        requestEncoder: RequestEncoder,
         urlSessionConfiguration: URLSessionConfiguration,
-        eventNotificationCenter: EventNotificationCenter
+        eventNotificationCenter: PersistentEventNotificationCenter
     ) -> WebSocketClient? {
         environment.webSocketClientBuilder?(
             urlSessionConfiguration,
-            requestEncoder,
             EventDecoder(),
             eventNotificationCenter
         )
@@ -114,7 +112,7 @@ class ChatClientFactory {
     func makeEventNotificationCenter(
         databaseContainer: DatabaseContainer,
         currentUserId: @escaping () -> UserId?
-    ) -> EventNotificationCenter {
+    ) -> PersistentEventNotificationCenter {
         let center = environment.notificationCenterBuilder(databaseContainer, nil)
         let middlewares: [EventMiddleware] = [
             EventDataProcessorMiddleware(),

Sources/StreamChat/Config/ChatClientConfig.swift

@@ -227,21 +227,3 @@ extension ChatClientConfig {
         public var latestMessagesLimit = 5
     }
 }
-
-/// A struct representing an API key of the chat app.
-///
-/// An API key can be obtained by registering on [our website](https://getstream.io/chat/trial/\).
-///
-public struct APIKey: Equatable, Sendable {
-    /// The string representation of the API key
-    public let apiKeyString: String
-
-    /// Creates a new `APIKey` from the provided string. Fails, if the string is empty.
-    ///
-    /// - Warning: The `apiKeyString` must be a non-empty value, otherwise an assertion failure is raised.
-    ///
-    public init(_ apiKeyString: String) {
-        log.assert(apiKeyString.isEmpty == false, "APIKey can't be initialize with an empty string.")
-        self.apiKeyString = apiKeyString
-    }
-}

Sources/StreamChat/Query/Filter.swift

@@ -592,14 +592,14 @@ private struct FilterRightSide: Decodable {
         self.operator = container.allKeys.first!.stringValue
         var value: FilterValue?
 
-        if let dateValue = try? container.decode(Date.self, forKey: key) {
-            value = dateValue
-        } else if let stringValue = try? container.decode(String.self, forKey: key) {
-            value = stringValue
-        } else if let intValue = try? container.decode(Int.self, forKey: key) {
+        if let intValue = try? container.decode(Int.self, forKey: key) {
             value = intValue
         } else if let doubleValue = try? container.decode(Double.self, forKey: key) {
             value = doubleValue
+        } else if let dateValue = try? container.decode(Date.self, forKey: key) {
+            value = dateValue
+        } else if let stringValue = try? container.decode(String.self, forKey: key) {
+            value = stringValue
         } else if let boolValue = try? container.decode(Bool.self, forKey: key) {
             value = boolValue
         } else if let stringArray = try? container.decode([String].self, forKey: key) {

Sources/StreamChat/Repositories/ConnectionRepository.swift

@@ -22,21 +22,25 @@ class ConnectionRepository: @unchecked Sendable {
         set { connectionQueue.async(flags: .barrier) { self._connectionId = newValue }}
     }
 
+    let webSocketConnectEndpoint = AllocatedUnfairLock<Endpoint<EmptyResponse>?>(nil)
     let isClientInActiveMode: Bool
     private let syncRepository: SyncRepository
+    private let webSocketEncoder: RequestEncoder?
     private let webSocketClient: WebSocketClient?
     private let apiClient: APIClient
     private let timerType: TimerScheduling.Type
 
     init(
         isClientInActiveMode: Bool,
         syncRepository: SyncRepository,
+        webSocketEncoder: RequestEncoder?,
         webSocketClient: WebSocketClient?,
         apiClient: APIClient,
         timerType: TimerScheduling.Type
     ) {
         self.isClientInActiveMode = isClientInActiveMode
         self.syncRepository = syncRepository
+        self.webSocketEncoder = webSocketEncoder
         self.webSocketClient = webSocketClient
         self.apiClient = apiClient
         self.timerType = timerType
@@ -80,6 +84,7 @@ class ConnectionRepository: @unchecked Sendable {
                 }
             }
         }
+        updateWebSocketConnectURLRequest()
         webSocketClient?.connect()
     }
 
@@ -114,29 +119,52 @@ class ConnectionRepository: @unchecked Sendable {
 
     /// Updates the WebSocket endpoint to use the passed token and user information for the connection
     func updateWebSocketEndpoint(with token: Token, userInfo: UserInfo?) {
-        webSocketClient?.connectEndpoint = .webSocketConnect(userInfo: userInfo ?? .init(id: token.userId))
+        webSocketConnectEndpoint.value = .webSocketConnect(userInfo: userInfo ?? .init(id: token.userId))
     }
-
+    
     /// Updates the WebSocket endpoint to use the passed user id
     func updateWebSocketEndpoint(with currentUserId: UserId) {
-        webSocketClient?.connectEndpoint = .webSocketConnect(userInfo: UserInfo(id: currentUserId))
+        webSocketConnectEndpoint.value = .webSocketConnect(userInfo: UserInfo(id: currentUserId))
+    }
+    
+    private func updateWebSocketConnectURLRequest() {
+        guard let webSocketClient, let webSocketEncoder, let webSocketConnectEndpoint = webSocketConnectEndpoint.value else { return }
+        let request: URLRequest? = {
+            do {
+                return try webSocketEncoder.encodeRequest(for: webSocketConnectEndpoint)
+            } catch {
+                log.error(error.localizedDescription, error: error)
+                return nil
+            }
+        }()
+        guard let request else { return }
+        webSocketClient.connectRequest = request
     }
 
     func handleConnectionUpdate(
         state: WebSocketConnectionState,
         onExpiredToken: () -> Void
     ) {
+        let event = ConnectionStatusUpdated(webSocketConnectionState: state)
+        if event.connectionStatus != connectionStatus {
+            // Publish Connection event with the new state
+            webSocketClient?.publishEvent(event)
+        }
+        
         connectionStatus = .init(webSocketConnectionState: state)
 
         // We should notify waiters if connectionId was obtained (i.e. state is .connected)
         // or for .disconnected state except for disconnect caused by an expired token
         let shouldNotifyConnectionIdWaiters: Bool
         let connectionId: String?
         switch state {
-        case let .connected(connectionId: id):
+        case let .connected(healthCheckInfo: healthCheckInfo):
             shouldNotifyConnectionIdWaiters = true
-            connectionId = id
-        case let .disconnected(source) where source.serverError?.isExpiredTokenError == true:
+            connectionId = healthCheckInfo.connectionId
+            syncRepository.syncLocalState {
+                log.info("Local state sync completed", subsystems: .offlineSupport)
+            }
+        case let .disconnected(source) where source.serverError?.isTokenExpiredError == true:
             onExpiredToken()
             shouldNotifyConnectionIdWaiters = false
             connectionId = nil
@@ -146,7 +174,7 @@ class ConnectionRepository: @unchecked Sendable {
         case .initialized,
              .connecting,
              .disconnecting,
-             .waitingForConnectionId:
+             .authenticating:
             shouldNotifyConnectionIdWaiters = false
             connectionId = nil
         }