fix superClient

Tun/Super/SDLSuperClient.swift

@@ -10,10 +10,10 @@ import Network
 
 final class SDLSuperClient: @unchecked Sendable {
     private let queue = DispatchQueue(label: "com.sdl.SuperClient.queue") // 专用队列保证线程安全
-    // 数据流
     public let messageStream: AsyncThrowingStream<SDLSuperMessage, Error>
-    private let messageCont: AsyncThrowingStream<SDLSuperMessage, Error>.Continuation
+    private let messageContinuation: AsyncThrowingStream<SDLSuperMessage, Error>.Continuation
     
+    private let readySignal = AsyncOneShot<Void>()
     private let stateLock = NSLock()
     private var isStarted = false
     private var isStopped = false
@@ -25,9 +25,9 @@ final class SDLSuperClient: @unchecked Sendable {
     init(serverEndpoint: SDLConfiguration.ResolvedServerEndpoint, port: UInt16, maxBufferSize: Int = 2 * 1024 * 1024) {
         self.maxBufferSize = maxBufferSize
         
-        let pairs = AsyncThrowingStream.makeStream(of: SDLSuperMessage.self)
+        let pairs = AsyncThrowingStream.makeStream(of: SDLSuperMessage.self, bufferingPolicy: .bufferingNewest(1024))
         self.messageStream = pairs.stream
-        self.messageCont = pairs.continuation
+        self.messageContinuation = pairs.continuation
         
         let options = NWProtocolTLS.Options()
         serverEndpoint.host.withCString {
@@ -61,18 +61,30 @@ final class SDLSuperClient: @unchecked Sendable {
         guard self.markStarted() else {
             return
         }
-
+        
-        let stateStream = self.makeStateStream()
         defer {
             self.stop()
         }
-
+        
-        try await withTaskCancellationHandler {
+        do {
-            self.connection.start(queue: self.queue)
+            try await withTaskCancellationHandler {
-            try await self.waitUntilReady(stateStream)
+                self.connection.stateUpdateHandler = { [weak self] state in
-            try await self.readLoop()
+                    self?.handleConnectionStateUpdate(state)
-        } onCancel: {
+                }
-            self.connection.cancel()
+                self.connection.start(queue: self.queue)
+                try await self.readySignal.wait()
+                try await self.readLoop()
+                self.finishMessageStream()
+            } onCancel: {
+                self.connection.stateUpdateHandler = nil
+                self.connection.cancel()
+            }
+        } catch is CancellationError {
+            self.finishMessageStream()
+            throw CancellationError()
+        } catch {
+            self.finishMessageStream(throwing: error)
+            throw error
         }
     }
 
@@ -88,36 +100,28 @@ final class SDLSuperClient: @unchecked Sendable {
         preconditionFailure("invalid super server IP: \(ip)")
     }
 
-    private func makeStateStream() -> AsyncThrowingStream<NWConnection.State, Error> {
+    private func handleConnectionStateUpdate(_ state: NWConnection.State) {
-        let connection = self.connection
+        SDLLogger.log("[SDLSuperClient] new state: \(state)", category: .super)
-        return AsyncThrowingStream(bufferingPolicy: .bufferingNewest(16)) { continuation in
+        switch state {
-            connection.stateUpdateHandler = { [weak self] state in
+        case .ready:
-                SDLLogger.log("[SDLSuperClient] new state: \(state)", category: .super)
+            Task {
-                switch state {
+                await self.readySignal.succeed(())
-                case .failed(let error):
-                    let wrappedError = SDLSuperError.connectionFailed(error)
-                    self?.finishMessageStream(throwing: wrappedError)
-                    continuation.finish(throwing: wrappedError)
-                case .cancelled:
-                    self?.finishMessageStream(throwing: SDLSuperError.connectionCancelled)
-                    continuation.finish(throwing: SDLSuperError.connectionCancelled)
-                default:
-                    continuation.yield(state)
-                }
             }
-        }
+        case .failed(let error):
-    }
+            let wrappedError = SDLSuperError.connectionFailed(error)
-
+            Task {
-    private func waitUntilReady(_ stateStream: AsyncThrowingStream<NWConnection.State, Error>) async throws {
+                await self.readySignal.fail(wrappedError)
-        for try await state in stateStream {
-            try Task.checkCancellation()
-
-            if case .ready = state {
-                return
             }
+            self.finishMessageStream(throwing: wrappedError)
+        case .cancelled:
+            let error = SDLSuperError.connectionCancelled
+            Task {
+                await self.readySignal.fail(error)
+            }
+            self.finishMessageStream(throwing: error)
+        default:
+            break
         }
-
-        throw SDLSuperError.connectionCancelled
     }
 
     private func readLoop() async throws {
@@ -129,7 +133,7 @@ final class SDLSuperClient: @unchecked Sendable {
             for frame in frames {
                 try Task.checkCancellation()
                 if let message = SDLSuperCodec.decode(frame: frame) {
-                    self.messageCont.yield(message)
+                    self.messageContinuation.yield(message)
                 } else {
                     throw SDLSuperError.decodeError("invalid message")
                 }
@@ -151,10 +155,11 @@ final class SDLSuperClient: @unchecked Sendable {
             if let error {
                 SDLLogger.log("[SDLSuperClient] send data get error: \(error)", category: .super)
                 self?.finishMessageStream(throwing: SDLSuperError.writeFailed(error))
+                self?.connection.cancel()
             }
         })
     }
-    
+
     private static func readOnce(connection: NWConnection) async throws -> Data {
         guard connection.state == .ready else {
             throw SDLSuperError.connectionCancelled
@@ -191,6 +196,9 @@ final class SDLSuperClient: @unchecked Sendable {
         let connection = self.connection
         connection.stateUpdateHandler = nil
         connection.cancel()
+        Task {
+            await self.readySignal.fail(SDLSuperError.connectionCancelled)
+        }
         self.finishMessageStream()
         
         SDLLogger.log("[SDLSuperClient] stopped", category: .super)
@@ -237,9 +245,9 @@ final class SDLSuperClient: @unchecked Sendable {
         }
 
         if let error {
-            self.messageCont.finish(throwing: error)
+            self.messageContinuation.finish(throwing: error)
         } else {
-            self.messageCont.finish()
+            self.messageContinuation.finish()
         }
     }
     

Tun/Super/SDLSuperSession.swift

@@ -20,20 +20,8 @@ final class SDLSuperSession: @unchecked Sendable {
         self.onMessage = onMessage
         self.client = SDLSuperClient(serverEndpoint: serverEndpoint, port: port)
     }
-
+    
     func run() async throws {
-        try await self.runLoops()
-    }
-
-    func stop() async {
-        self.client.stop()
-    }
-
-    func send(type: SDLPacketType, data: Data) async {
-        self.client.send(type: type, data: data)
-    }
-
-    private func runLoops() async throws {
         SDLLogger.log("[SDLSuperSession] start super client: \(self.serverEndpoint.ip)", category: .super)
 
         try await withThrowingTaskGroup(of: Void.self) { group in
@@ -56,6 +44,14 @@ final class SDLSuperSession: @unchecked Sendable {
             _ = try await group.next()
         }
     }
+    
+    func stop() async {
+        self.client.stop()
+    }
+
+    func send(type: SDLPacketType, data: Data) async {
+        self.client.send(type: type, data: data)
+    }
 
     private func readLoop() async throws {
         for try await message in self.client.messageStream {