anlicheng/punchnet-macos
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix quicClient

Browse Source
This commit is contained in:
anlicheng 2026-04-28 17:19:55 +08:00
parent 7b549981af
commit 2b234fb5c3
2 changed files with 82 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

122
Tun/Punchnet/Actors/SDLContextActor.swift
View File

@ -258,76 +258,94 @@ actor SDLContextActor {
private func startQUICClient() async throws { private func startQUICClient() async throws {
SDLLogger.log("[SDLContext] try start quicClient", for: .debug) SDLLogger.log("[SDLContext] try start quicClient", for: .debug)
self.quicWorker?.cancel() self.quicWorker?.cancel()
await self.quicClient?.stop() await self.quicClient?.stop()
// 启动monitor // 启动monitor
let quicClient = SDLQUICClient(host: self.config.serverHost, port: 443) let quicClient = SDLQUICClient(host: self.config.serverHost, port: 443)
self.quicClient = quicClient self.quicClient = quicClient
await quicClient.start() await quicClient.start()
try await quicClient.waitReady(timeout: .seconds(3)) do {
// 这里必须等待quic的协商完成 try await quicClient.waitReady(timeout: .seconds(3))
try await Task.sleep(for: .seconds(0.3)) // 这里必须等待quic的协商完成
SDLLogger.log("[SDLContext] start quic client: \(self.config.serverHost)") 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 try await withTaskCancellationHandler {
defer { try await withThrowingTaskGroup { group in
group.cancelAll() defer {
} group.cancelAll()
}
group.addTask {
for await message in await quicClient.messageStream { group.addTask {
for await message in await quicClient.messageStream {
if Task.isCancelled {
return
}
await self.handleQUICMessage(message: message)
}
if Task.isCancelled { if Task.isCancelled {
return return
} }
await self.handleQUICMessage(message: message) throw SDLQUICClientExit.transportClosed("messageStream finished")
} }
throw SDLQUICClientExit.transportClosed("messageStream finished")
} group.addTask {
let exit = await quicClient.run()
group.addTask {
let exit = await quicClient.run() switch exit {
case .normal:
switch exit {
case .normal:
return
case .cancelled:
if Task.isCancelled {
return 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: case .cancelled:
throw SDLQUICClientExit.cancelled if Task.isCancelled {
case .writeFailed(let error): return
throw error }
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
}
}
if Task.isCancelled {
return
}
throw SDLQUICClientExit.transportClosed("eventStream finished")
}
do {
let _ = try await group.next()
await quicClient.stop()
} catch {
await quicClient.stop()
throw error
}
}
} onCancel: {
Task {
await quicClient.stop()
} }
let _ = try await group.next()
}
} onCancel: {
Task {
await quicClient.stop()
} }
} catch {
await quicClient.stop()
throw error
} }
} }
private func handleQUICMessage(message: SDLQUICInboundMessage) async { private func handleQUICMessage(message: SDLQUICInboundMessage) async {
switch message { switch message {
case .welcome(let welcome): case .welcome(let welcome):

31
Tun/Punchnet/Actors/SDLQuicClient.swift
View File

@ -136,27 +136,20 @@ actor SDLQUICClient {
} }
func run() async -> SDLQUICClientExit { func run() async -> SDLQUICClientExit {
await withTaskCancellationHandler { await withTaskGroup(of: SDLQUICClientExit.self) { group in
await withTaskGroup(of: SDLQUICClientExit.self) { group in group.addTask {
group.addTask { await self.readLoop()
await self.readLoop()
}
group.addTask {
await self.heartbeatLoop()
}
let exit = await group.next() ?? .normal
group.cancelAll()
await self.stop()
self.finishStreams()
return exit
} }
} onCancel: {
Task { group.addTask {
await self.stop() await self.heartbeatLoop()
} }
let exit = await group.next() ?? .normal
group.cancelAll()
self.finishStreams()
return exit
} }
} }