diff --git a/Tun/Punchnet/Actors/SDLContextActor.swift b/Tun/Punchnet/Actors/SDLContextActor.swift index 89c298e..7b1b1e7 100644 --- a/Tun/Punchnet/Actors/SDLContextActor.swift +++ b/Tun/Punchnet/Actors/SDLContextActor.swift @@ -165,67 +165,7 @@ actor SDLContextActor { await self.startDnsLocalClient() await self.supervisor.addWorker(name: "quicClient") { - SDLLogger.log("[SDLContext] try start quicClient", for: .debug) - - self.quicWorker?.cancel() - await self.quicClient?.stop() - - // 启动monitor - let quicClient = SDLQUICClient(host: self.config.serverHost, port: 443) - self.quicClient = quicClient - - await quicClient.start() - - try await quicClient.waitReady(timeout: .seconds(3)) - // 这里必须等待quic的协商完成 - try await Task.sleep(for: .seconds(0.3)) - SDLLogger.log("[SDLContext] start quic client: \(self.config.serverHost)") - - try await withThrowingTaskGroup { group in - defer { - group.cancelAll() - } - - group.addTask { - for await message in await quicClient.messageStream { - await self.handleQUICMessage(message: message) - } - } - - group.addTask { - let exit = await quicClient.run() - - switch exit { - case .normal: - return - case .cancelled: - if Task.isCancelled { - return - } - throw exit - case .transportClosed, .readFailed, .writeFailed: - throw exit - } - } - - group.addTask { - for await event in await quicClient.eventStream { - switch event { - case .failed(let error): - throw error - case .cancelled: - throw SDLQUICClientExit.cancelled - case .writeFailed(let error): - throw error - } - } - } - - while let _ = try await group.next() { - () - } - } - + try await self.startQUICClient() } await self.supervisor.addWorker(name: "udpHole") { @@ -316,6 +256,78 @@ actor SDLContextActor { try await self.setNetworkSettings(config: config, dnsServer: DNSHelper.dnsServer) } + private func startQUICClient() async throws { + SDLLogger.log("[SDLContext] try start quicClient", for: .debug) + + self.quicWorker?.cancel() + await self.quicClient?.stop() + + // 启动monitor + let quicClient = SDLQUICClient(host: self.config.serverHost, port: 443) + self.quicClient = quicClient + + await quicClient.start() + + try await quicClient.waitReady(timeout: .seconds(3)) + // 这里必须等待quic的协商完成 + try await Task.sleep(for: .seconds(0.3)) + SDLLogger.log("[SDLContext] start quic client: \(self.config.serverHost)") + + try await withTaskCancellationHandler{ + try await withThrowingTaskGroup { group in + defer { + group.cancelAll() + } + + group.addTask { + for await message in await quicClient.messageStream { + if Task.isCancelled { + return + } + await self.handleQUICMessage(message: message) + } + throw SDLQUICClientExit.transportClosed("messageStream finished") + } + + group.addTask { + let exit = await quicClient.run() + + switch exit { + case .normal: + return + case .cancelled: + if Task.isCancelled { + return + } + throw exit + case .transportClosed, .readFailed, .writeFailed: + throw exit + } + } + + group.addTask { + for await event in await quicClient.eventStream { + switch event { + case .failed(let error): + throw error + case .cancelled: + throw SDLQUICClientExit.cancelled + case .writeFailed(let error): + throw error + } + } + } + + let _ = try await group.next() + } + + } onCancel: { + Task { + await quicClient.stop() + } + } + } + private func handleQUICMessage(message: SDLQUICInboundMessage) async { switch message { case .welcome(let welcome): diff --git a/Tun/Punchnet/Actors/SDLQuicClient.swift b/Tun/Punchnet/Actors/SDLQuicClient.swift index d2881b4..58e5c17 100644 --- a/Tun/Punchnet/Actors/SDLQuicClient.swift +++ b/Tun/Punchnet/Actors/SDLQuicClient.swift @@ -79,7 +79,7 @@ actor SDLQUICClient { (self.messageStream, self.messageCont) = AsyncStream.makeStream(of: SDLQUICInboundMessage.self) (self.eventStream, self.eventCont) = AsyncStream.makeStream(of: SDLQUICEvent.self) - // TODO 这里设置证书的校验逻辑 + // 这里设置证书的校验逻辑 sec_protocol_options_set_verify_block( options.securityProtocolOptions, { metadata, trust, complete in @@ -130,16 +130,16 @@ actor SDLQUICClient { group.addTask { await self.readLoop() } - + group.addTask { await self.heartbeatLoop() } - + let exit = await group.next() ?? .normal group.cancelAll() await self.stop() self.finishStreams() - + return exit } } onCancel: { @@ -204,11 +204,7 @@ actor SDLQUICClient { self.messageCont.finish() self.eventCont.finish() } - - deinit { - self.connection.cancel() - self.finishStreams() - } + } // --MARK: Ready状态机 @@ -233,7 +229,7 @@ extension SDLQUICClient { if Task.isCancelled { return } - await self.cancelWaiter(id: id, throwing: SDLQUICError.timeout) + self.cancelWaiter(id: id, throwing: SDLQUICError.timeout) } defer {