Tun/Punchnet/Actors/ArpServer.swift

@@ -82,7 +82,7 @@ actor ArpServer {
         var arpRequest = SDLArpRequest()
         arpRequest.targetIp = targetIp
         
-        await quicClient.send(type: .arpRequest, data: try arpRequest.serializedData())
+        quicClient.send(type: .arpRequest, data: try arpRequest.serializedData())
     }
     
     func handleArpResponse(arpResponse: SDLArpResponse) {

Tun/Punchnet/Actors/SDLContextActor.swift

@@ -166,48 +166,71 @@ actor SDLContextActor {
         // 启动monitor
         let quicClient = SDLQUICClient(host: self.config.serverHost, port: 1443)
         self.quicClient = quicClient
-        await quicClient.start()
+        quicClient.start()
         
         defer {
-            Task {
+            self.quicClient?.stop()
-                await self.quicClient?.stop()
+            self.quicClient = nil
-                self.quicClient = nil
+            SDLLogger.log("[SDLContext] quicClient: stop")
-                SDLLogger.log("[SDLContext] quicClient: stop")
-            }
         }
         
         // 这里必须等待quic的协商完成
         try await Task.sleep(for: .seconds(0.5))
         SDLLogger.log("[SDLContext] start quic client: \(self.config.serverHost)")
         
-        try await withTaskCancellationHandler {
+        try await withThrowingTaskGroup { group in
-            try await withThrowingTaskGroup { group in
+            defer {
-                defer {
+                group.cancelAll()
-                    group.cancelAll()
+            }
-                }
             
-                group.addTask {
+            // 创建一个简单的异步状态等待机制（可以用一个 Actor 或者 AsyncStream 模拟）
-                    for try await message in await quicClient.messageStream {
+            let (readyStream, readyContinuation) = AsyncStream<Void>.makeStream()
-                        try Task.checkCancellation()
+            
-                        await self.handleQUICMessage(message: message)
+            group.addTask {
+                for await event in quicClient.eventStream {
+                    try Task.checkCancellation()
+                    switch event {
+                    case .ready:
+                        readyContinuation.yield()
+                    case .failed(let error):
+                        throw error
+                    case .cancelled:
+                        throw SDLQUICEvent.cancelled
+                    case .writeFailed(let error):
+                        throw error
                     }
                 }
+            }
             
-                group.addTask {
+            group.addTask {
-                    while true {
+                // 等待信号
-                        try await Task.sleep(for: .seconds(5))
+                var it = readyStream.makeAsyncIterator()
-                        try Task.checkCancellation()
+                await it.next()
-                        await quicClient.send(type: .ping, data: Data())
+                
+                try Task.checkCancellation()
+                
+                await withThrowingTaskGroup { workerGroup in
+                    workerGroup.addTask {
+                        for await message in quicClient.messageStream {
+                            try Task.checkCancellation()
+                            await self.handleQUICMessage(message: message)
+                        }
                     }
-                    SDLLogger.log("[SDLQUICClient] udp pingTask cancel", for: .debug)
-                }
                     
-                try await group.next()
+                    workerGroup.addTask {
-            }
+                        let timerStream = SDLAsyncTimerStream()
-        } onCancel: {
+                        timerStream.start(interval: .seconds(5))
-            Task {
+                        
-                await quicClient.stop()
+                        for await _ in timerStream.stream {
+                            try Task.checkCancellation()
+                            quicClient.send(type: .ping, data: Data())
+                        }
+                        SDLLogger.log("[SDLQUICClient] udp pingTask cancel", for: .debug)
+                    }
+                }
             }
+            
+            try await group.next()
         }
     }
     
@@ -857,7 +880,7 @@ extension SDLContextActor {
         
         if let registerSuperData = try? registerSuper.serializedData() {
             SDLLogger.log("[SDLContext] will send register super")
-            await self.quicClient?.send(type: .registerSuper, data: registerSuperData)
+            self.quicClient?.send(type: .registerSuper, data: registerSuperData)
         }
     }
     

Tun/Punchnet/Actors/SDLPuncherActor.swift

@@ -90,7 +90,7 @@ actor SDLPuncherActor {
             phase: .waitingPeerInfo(deadline: now.addingTimeInterval(self.peerInfoTimeout))
         )
         
-        await quicClient.send(type: .queryInfo, data: queryData)
+        quicClient.send(type: .queryInfo, data: queryData)
     }
     
     func handlePeerInfo(using udpHole: SDLUDPHole?, udpHoleV6: SDLUDPHoleV6?, peerInfo: SDLPeerInfo) async {

Tun/Punchnet/Actors/SDLQuicClient.swift

@@ -15,61 +15,55 @@ import Security
 enum SDLQUICError: Error {
     case connectionFailed(Error)
     case connectionCancelled
-    case writeFailed(Error)
-    
-    case internalError(Error)
-    
     case timeout
     case decodeError(String)
     case packetTooLarge
     case dataStreamClosed
 }
 
-actor SDLQUICClient {
+enum SDLQUICEvent: Error {
-    enum State {
+    case ready
-        case idle
+    case failed(Error)
-        case running
+    case cancelled
-        case stopped
+    case writeFailed(Error)
-    }
+}
-    
-    private var state: State = .idle
 
+final class SDLQUICClient {
     private let frameParser: SDLQUICFrameParser
     
     // 数据流
-    public var messageStream: AsyncThrowingStream<SDLQUICInboundMessage, Error>
+    public var messageStream: AsyncStream<SDLQUICInboundMessage>
-    private let messageCont: AsyncThrowingStream<SDLQUICInboundMessage, Error>.Continuation
+    private let messageCont: AsyncStream<SDLQUICInboundMessage>.Continuation
-    private var isMessageContinuationFinished: Bool = false
-    
     private var readTask: Task<Void, Never>?
 
-    private var connection: NWConnection?
+    // 事件流
+    public var eventStream: AsyncStream<SDLQUICEvent>
+    private let eventCont: AsyncStream<SDLQUICEvent>.Continuation
+    
+    private var isFinished: Bool = false
+
+    private let connection: NWConnection
     private let queue = DispatchQueue(label: "com.sdl.QUICClient.queue") // 专用队列保证线程安全
 
-    private let host: String
-    private let port: UInt16
-    
     init(host: String, port: UInt16, maxBufferSize: Int = 2 * 1024 * 1024) {
-        self.host = host
-        self.port = port
-        
-        self.frameParser = SDLQUICFrameParser(maxBufferSize: maxBufferSize)
-        (self.messageStream, self.messageCont) = AsyncThrowingStream.makeStream(of: SDLQUICInboundMessage.self)
-    }
-    
-    func start() {
         let options = NWProtocolTLS.Options()
+        
         sec_protocol_options_add_tls_application_protocol(
             options.securityProtocolOptions,
             "punchnet/1.0"
         )
         
+        self.frameParser = SDLQUICFrameParser(maxBufferSize: maxBufferSize)
+        
+        (self.eventStream, self.eventCont) = AsyncStream.makeStream(of: SDLQUICEvent.self)
+        (self.messageStream, self.messageCont) = AsyncStream.makeStream(of: SDLQUICInboundMessage.self)
+
         // 这里设置证书的校验逻辑
         sec_protocol_options_set_verify_block(
             options.securityProtocolOptions,
             { metadata, trust, complete in
                 // 执行公钥校验
-                complete(TLSVerifier.verify(trust: trust, host: self.host))
+                complete(TLSVerifier.verify(trust: trust, host: host))
             },
             self.queue
         )
@@ -79,95 +73,64 @@ actor SDLQUICClient {
         // 关键：让 Network.framework 忽略系统代理
         params.preferNoProxies = true
         
-        let connection = NWConnection(host: .init(host), port: .init(rawValue: port)!, using: params)
+        self.connection = NWConnection(host: .init(host), port: .init(rawValue: port)!, using: params)
+    }
     
+    func start() {
         connection.stateUpdateHandler = { [weak self] state in
             SDLLogger.log("[SDLQUICClient] new state: \(state)", for: .debug)
-            Task {
+            switch state {
-                await self?.handleConnectionState(state: state)
+            case .ready:
+                self?.startReadTask()
+                self?.eventCont.yield(.ready)
+            case .failed(let error):
+                self?.eventCont.yield(.failed(error))
+            case .cancelled:
+                self?.eventCont.yield(.cancelled)
+            default:
+                ()
             }
         }
         connection.start(queue: self.queue)
-        
-        self.connection = connection
-    }
-    
-    private func handleConnectionState(state: NWConnection.State) {
-        switch state {
-        case .ready:
-            self.startReadTask()
-            self.state = .running
-        case .failed(let error):
-            self.finishMessageContinuationIfNeed(throwing: .connectionFailed(error))
-        case .cancelled:
-            self.finishMessageContinuationIfNeed(throwing: .connectionCancelled)
-        default:
-            ()
-        }
-    }
-    
-    private func finishMessageContinuationIfNeed(throwing error: SDLQUICError?) {
-        guard !self.isMessageContinuationFinished else {
-            return
-        }
-        
-        self.isMessageContinuationFinished = true
-        if let error {
-            self.messageCont.finish(throwing: error)
-        } else {
-            self.messageCont.finish()
-        }
     }
     
     private func startReadTask() {
         self.readTask?.cancel()
         self.readTask = Task {
             do {
-                while true {
+                while !Task.isCancelled {
-                    try Task.checkCancellation()
                     let data = try await self.readOnce()
                     let frames = try self.frameParser.parseFrames(data: data)
                     for frame in frames {
                         if let message = SDLQUICCodec.decode(frame: frame) {
                             self.messageCont.yield(message)
-                        } else {
-                            self.finishMessageContinuationIfNeed(throwing: .decodeError("invalid message"))
                         }
                     }
                 }
-            } catch let err {
+            } catch {
-                self.finishMessageContinuationIfNeed(throwing: .internalError(err))
+                self.messageCont.finish()
             }
         }
     }
     
     func send(type: SDLPacketType, data: Data) {
-        guard case .running = state, let connection = self.connection, connection.state == .ready else {
-            return
-        }
-        
         var len = UInt16(data.count + 1).bigEndian
+
         var packet = Data(Data(bytes: &len, count: 2))
         packet.append(type.rawValue)
         packet.append(data)
 
         connection.send(content: packet, completion: .contentProcessed { [weak self] error in
             if let error {
-                Task {
+                SDLLogger.log("[SDLQUICClient] send data get error: \(error)", for: .debug)
-                    SDLLogger.log("[SDLQUICClient] send data get error: \(error)", for: .debug)
+                self?.eventCont.yield(.writeFailed(error))
-                    await self?.finishMessageContinuationIfNeed(throwing: .writeFailed(error))
-                }
             }
         })
     }
     
     private func readOnce() async throws -> Data {
-        guard let connection = self.connection else {
-            throw SDLQUICError.connectionCancelled
-        }
-        
         return try await withCheckedThrowingContinuation { cont in
-            connection.receive(minimumIncompleteLength: 1, maximumLength: 64 * 1024) { data, _, isComplete, error in
+            self.connection.receive(minimumIncompleteLength: 1, maximumLength: 64 * 1024) { data, _, isComplete, error in
                 if let error {
                     cont.resume(throwing: error)
                     return
@@ -183,15 +146,10 @@ actor SDLQUICClient {
     }
 
     func stop() {
-        guard self.state != .stopped else {
-            return
-        }
-        
-        self.state = .stopped
-        
         self.readTask?.cancel()
-        self.connection?.cancel()
+        self.connection.cancel()
-        self.finishMessageContinuationIfNeed(throwing: nil)
+        self.eventCont.finish()
+        self.messageCont.finish()
     }
     
 }