refactor: integrate new incremental sync - WPB-15440

WireAPI/Sources/WireAPI/Network/NetworkService/NetworkService.swift

@@ -111,8 +111,10 @@ extension NetworkService: URLSessionWebSocketDelegate {
         didCloseWith closeCode: URLSessionWebSocketTask.CloseCode,
         reason: Data?
     ) {
-        webSocketsByTask[webSocketTask]?.close()
-        webSocketsByTask[webSocketTask] = nil
+        Task {
+            await webSocketsByTask[webSocketTask]?.close()
+            webSocketsByTask[webSocketTask] = nil
+        }
     }
 
 }

WireAPI/Sources/WireAPI/Network/PushChannel/PushChannel.swift

@@ -19,20 +19,20 @@
 import Foundation
 import WireFoundation
 
-final class PushChannel: PushChannelProtocol {
+public final class PushChannel: PushChannelProtocol {
 
-    typealias Stream = AsyncThrowingStream<UpdateEventEnvelope, any Error>
+    public typealias Stream = AsyncThrowingStream<UpdateEventEnvelope, any Error>
 
     private let webSocket: any WebSocketProtocol
     private let decoder = JSONDecoder()
 
-    init(webSocket: any WebSocketProtocol) {
+    public init(webSocket: any WebSocketProtocol) {
         self.webSocket = webSocket
     }
 
-    func open() throws -> Stream {
+    public func open() async throws -> Stream {
         print("opening new push channel")
-        return try webSocket.open().map { [weak self, decoder] message in
+        return try await webSocket.open().map { [weak self, decoder] message in
             do {
                 switch message {
                 case let .data(data):
@@ -50,15 +50,15 @@ final class PushChannel: PushChannelProtocol {
                 }
             } catch {
                 print("failed to get next web socket message: \(error)")
-                self?.close()
+                await self?.close()
                 throw error
             }
         }.toStream()
     }
 
-    func close() {
+    public func close() async {
         print("closing push channel")
-        webSocket.close()
+        await webSocket.close()
     }
 
 }

WireAPI/Sources/WireAPI/Network/PushChannel/PushChannelProtocol.swift

@@ -20,16 +20,16 @@ import Foundation
 
 // sourcery: AutoMockable
 /// Make a direct connection to a server to receive update events.
-public protocol PushChannelProtocol {
+public protocol PushChannelProtocol: Sendable {
 
     /// Open the push channel and start receiving update events.
     ///
     /// - Returns: An async stream of live update event envelopes.
 
-    func open() throws -> AsyncThrowingStream<UpdateEventEnvelope, any Error>
+    func open() async throws -> AsyncThrowingStream<UpdateEventEnvelope, any Error>
 
     /// Close the push channel and stop receiving update events.
 
-    func close()
+    func close() async
 
 }

WireAPI/Sources/WireAPI/Network/PushChannel/PushChannelService.swift

@@ -37,7 +37,7 @@ public final class PushChannelService: PushChannelServiceProtocol {
     private let networkService: NetworkService
     private let authenticationManager: any AuthenticationManagerProtocol
 
-    init(
+    public init(
         networkService: NetworkService,
         authenticationManager: any AuthenticationManagerProtocol
     ) {

WireAPI/Sources/WireAPI/Network/PushChannel/URLSessionWebSocketTaskProtocol.swift

@@ -19,7 +19,7 @@
 import Foundation
 
 // sourcery: AutoMockable
-protocol URLSessionWebSocketTaskProtocol {
+public protocol URLSessionWebSocketTaskProtocol: Sendable {
 
     var isOpen: Bool { get }
 
@@ -30,15 +30,15 @@ protocol URLSessionWebSocketTaskProtocol {
         reason: Data?
     )
 
-    func receive(completionHandler: @escaping (Result<URLSessionWebSocketTask.Message, any Error>) -> Void)
+    func receive(completionHandler: @escaping @Sendable (Result<URLSessionWebSocketTask.Message, any Error>) -> Void)
 
     func receive() async throws -> URLSessionWebSocketTask.Message
 
 }
 
 extension URLSessionWebSocketTask: URLSessionWebSocketTaskProtocol {
 
-    var isOpen: Bool {
+    public var isOpen: Bool {
         closeCode == .invalid
     }
 

WireAPI/Sources/WireAPI/Network/PushChannel/WebSocket.swift

@@ -18,22 +18,18 @@
 
 import Foundation
 
-final class WebSocket: WebSocketProtocol {
+public actor WebSocket: WebSocketProtocol {
 
-    typealias Stream = AsyncThrowingStream<URLSessionWebSocketTask.Message, any Error>
+    public typealias Stream = AsyncThrowingStream<URLSessionWebSocketTask.Message, any Error>
 
     private let connection: any URLSessionWebSocketTaskProtocol
     private var continuation: Stream.Continuation?
 
-    init(connection: any URLSessionWebSocketTaskProtocol) {
+    public init(connection: any URLSessionWebSocketTaskProtocol) {
         self.connection = connection
     }
 
-    deinit {
-        close()
-    }
-
-    func open() throws -> Stream {
+    public func open() async throws -> Stream {
         connection.resume()
 
         if #available(iOS 17, *) {
@@ -62,7 +58,7 @@ final class WebSocket: WebSocketProtocol {
             return Stream { continuation in
                 self.continuation = continuation
 
-                func yieldNextMessage() {
+                @Sendable func yieldNextMessage() {
                     guard connection.isOpen else {
                         continuation.finish()
                         return
@@ -85,7 +81,7 @@ final class WebSocket: WebSocketProtocol {
         }
     }
 
-    func close() {
+    public func close() async {
         connection.cancel(with: .goingAway, reason: nil)
         continuation?.finish()
     }

WireAPI/Sources/WireAPI/Network/PushChannel/WebSocketProtocol.swift

@@ -19,10 +19,10 @@
 import Foundation
 
 // sourcery: AutoMockable
-protocol WebSocketProtocol {
+public protocol WebSocketProtocol: Sendable {
 
-    func open() throws -> AsyncThrowingStream<URLSessionWebSocketTask.Message, any Error>
+    func open() async throws -> AsyncThrowingStream<URLSessionWebSocketTask.Message, any Error>
 
-    func close()
+    func close() async
 
 }

WireAPI/Tests/WireAPITests/Network/PushChannel/PushChannelTests.swift

@@ -57,7 +57,7 @@ final class PushChannelTests: XCTestCase {
         }
 
         // When the push channel is open and the stream is iterated
-        let liveEventEnvelopes = try sut.open()
+        let liveEventEnvelopes = try await sut.open()
 
         var receivedEnvelopes = [UpdateEventEnvelope]()
         for try await envelope in liveEventEnvelopes {
@@ -74,10 +74,10 @@ final class PushChannelTests: XCTestCase {
     func testClosingPushChannel() async throws {
         // Given an open push channel
         webSocket.open_MockValue = AsyncThrowingStream { _ in }
-        _ = try sut.open()
+        _ = try await sut.open()
 
         // When the push channel is closed
-        sut.close()
+        await sut.close()
 
         // Then the web socket was closed
         XCTAssertEqual(webSocket.close_Invocations.count, 1)
@@ -91,7 +91,7 @@ final class PushChannelTests: XCTestCase {
             // Don't call finish, so the stream stays open.
         }
 
-        let liveEventEnvelopes = try sut.open()
+        let liveEventEnvelopes = try await sut.open()
 
         do {
             for try await _ in liveEventEnvelopes {
@@ -115,7 +115,7 @@ final class PushChannelTests: XCTestCase {
             // Don't call finish, so the stream stays open.
         }
 
-        let liveEventEnvelopes = try sut.open()
+        let liveEventEnvelopes = try await sut.open()
 
         do {
             for try await _ in liveEventEnvelopes {

WireAPI/Tests/WireAPITests/Network/PushChannel/WebSocketTests.swift

@@ -37,19 +37,6 @@ final class WebSocketTests: XCTestCase {
         try await super.tearDown()
     }
 
-    func testWebSocketCancelsWhenItDeinitializes() async throws {
-        // When
-        {
-            _ = WebSocket(connection: self.connection)
-        }()
-
-        // Then
-        let invocations = connection.cancelWithReason_Invocations
-        try XCTAssertCount(invocations, count: 1)
-        XCTAssertEqual(invocations[0].closeCode, .goingAway)
-        XCTAssertNil(invocations[0].reason)
-    }
-
     func testWebSocketCloses() async throws {
         // Given we're iterating over the web socket
         let sut = WebSocket(connection: connection)
@@ -69,7 +56,7 @@ final class WebSocketTests: XCTestCase {
 
         Task {
             do {
-                for try await _ in try sut.open() {
+                for try await _ in try await sut.open() {
                     didReceiveMessage.fulfill()
                 }
             } catch {
@@ -83,7 +70,7 @@ final class WebSocketTests: XCTestCase {
         await fulfillment(of: [didReceiveMessage], timeout: 1)
 
         // When
-        sut.close()
+        await sut.close()
 
         // Then the stream finished successfully
         await fulfillment(of: [didFinishIterating], timeout: 1)
@@ -113,7 +100,7 @@ final class WebSocketTests: XCTestCase {
 
         Task {
             do {
-                for try await _ in try sut.open() {
+                for try await _ in try await sut.open() {
                     didReceiveMessage.fulfill()
                 }
             } catch {
@@ -157,7 +144,7 @@ final class WebSocketTests: XCTestCase {
 
         Task {
             do {
-                for try await _ in try sut.open() {
+                for try await _ in try await sut.open() {
                     didReceiveMessage.fulfill()
                 }
             } catch {
@@ -199,7 +186,7 @@ final class WebSocketTests: XCTestCase {
 
         // When
         do {
-            for try await message in try sut.open() {
+            for try await message in try await sut.open() {
                 if case let .data(data) = message {
                     receivedMessageData.append(data)
                 }

WireDomain/Sources/WireDomain/Repositories/UpdateEvents/UpdateEventsRepositoryError.swift → WireDomain/Sources/WireDomain/Components/ClientSessionComponent+IncrementalSyncProvider.swift

@@ -18,9 +18,10 @@
 
 import Foundation
 
-enum UpdateEventsRepositoryError: Error {
+extension ClientSessionComponent: IncrementalSyncProvider {
 
-    case lastEventIDMissing
-    case failedToDecodeStoredEvent(Error)
+    public func provideIncrementalSync() throws -> IncrementalSync {
+        incrementalSync
+    }
 
 }

WireDomain/Sources/WireDomain/Components/ClientSessionComponent+InitialSyncBuilder.swift → WireDomain/Sources/WireDomain/Components/ClientSessionComponent+InitialSyncProvider.swift

@@ -18,9 +18,9 @@
 
 import Foundation
 
-extension ClientSessionComponent: InitialSyncBuilderProtocol {
+extension ClientSessionComponent: InitialSyncProvider {
 
-    public func buildInitialSync() throws -> any InitialSyncProtocol {
+    public func provideInitialSync() throws -> any InitialSyncProtocol {
         initialSync
     }