fix superClient
This commit is contained in:
parent
5e8b403fbf
commit
4d9f6b9a42
@ -109,7 +109,7 @@ actor PacketOutboundActor {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func stop() async {
|
func stop() {
|
||||||
self.packetReaderGeneration &+= 1
|
self.packetReaderGeneration &+= 1
|
||||||
|
|
||||||
let packetReaderTask = self.packetReaderTask
|
let packetReaderTask = self.packetReaderTask
|
||||||
@ -125,10 +125,6 @@ actor PacketOutboundActor {
|
|||||||
self.policyService.recordOutboundFlow(ipPacket: packet)
|
self.policyService.recordOutboundFlow(ipPacket: packet)
|
||||||
}
|
}
|
||||||
|
|
||||||
await self.handleTunRouteDecision(decision)
|
|
||||||
}
|
|
||||||
|
|
||||||
private func handleTunRouteDecision(_ decision: PacketOutboundRouter.RouteDecision) async {
|
|
||||||
switch decision {
|
switch decision {
|
||||||
case .loopback(let ipPacketData):
|
case .loopback(let ipPacketData):
|
||||||
let nePacket = NEPacket(data: ipPacketData, protocolFamily: 2)
|
let nePacket = NEPacket(data: ipPacketData, protocolFamily: 2)
|
||||||
@ -174,12 +170,12 @@ actor PacketOutboundActor {
|
|||||||
case .superNode(let payload):
|
case .superNode(let payload):
|
||||||
await self.sendSuperPacket(type: .data, data: payload)
|
await self.sendSuperPacket(type: .data, data: payload)
|
||||||
case .peer(let payload, let session):
|
case .peer(let payload, let session):
|
||||||
SDLLogger.log("[PacketOutboundActor] step 5 send packet by session: \(session)", for: .trace)
|
SDLLogger.log("[PacketOutboundActor] send packet by session: \(session)", for: .trace)
|
||||||
await self.sendPeerPacket(type: .data, data: payload, remoteAddress: session.natAddress)
|
await self.sendPeerPacket(type: .data, data: payload, remoteAddress: session.natAddress)
|
||||||
self.flowTracer.inc(num: payload.count, type: .p2p)
|
self.flowTracer.inc(num: payload.count, type: .p2p)
|
||||||
case .superNodeAndPunch(let payload, let request):
|
case .superNodeAndPunch(let payload, let request):
|
||||||
await self.sendSuperPacket(type: .data, data: payload)
|
await self.sendSuperPacket(type: .data, data: payload)
|
||||||
SDLLogger.log("[PacketOutboundActor] step 5 send packet by super: \(self.stunSocketAddress)", for: .trace)
|
SDLLogger.log("[PacketOutboundActor] send packet by super: \(self.stunSocketAddress)", for: .trace)
|
||||||
self.flowTracer.inc(num: payload.count, type: .forward)
|
self.flowTracer.inc(num: payload.count, type: .forward)
|
||||||
|
|
||||||
if let queryData = await self.puncherActor.makeQueryInfoRequest(request: request) {
|
if let queryData = await self.puncherActor.makeQueryInfoRequest(request: request) {
|
||||||
|
|||||||
@ -145,6 +145,7 @@ actor SDLSuperClient {
|
|||||||
let data = try await self.readOnce()
|
let data = try await self.readOnce()
|
||||||
let frames = try self.frameParser.parseFrames(data: data)
|
let frames = try self.frameParser.parseFrames(data: data)
|
||||||
for frame in frames {
|
for frame in frames {
|
||||||
|
try Task.checkCancellation()
|
||||||
if let message = SDLQUICCodec.decode(frame: frame) {
|
if let message = SDLQUICCodec.decode(frame: frame) {
|
||||||
self.messageCont.yield(message)
|
self.messageCont.yield(message)
|
||||||
} else {
|
} else {
|
||||||
@ -183,19 +184,26 @@ actor SDLSuperClient {
|
|||||||
throw SDLQUICError.connectionCancelled
|
throw SDLQUICError.connectionCancelled
|
||||||
}
|
}
|
||||||
|
|
||||||
return try await withCheckedThrowingContinuation { cont in
|
let readContinuation = OnceContinuation<Data, Error>()
|
||||||
connection.receive(minimumIncompleteLength: 1, maximumLength: 64 * 1024) { data, _, isComplete, error in
|
|
||||||
if let error {
|
|
||||||
cont.resume(throwing: error)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
if isComplete {
|
return try await withTaskCancellationHandler {
|
||||||
cont.resume(throwing: SDLQUICError.dataStreamClosed)
|
try await withCheckedThrowingContinuation { cont in
|
||||||
} else {
|
readContinuation.set(cont)
|
||||||
cont.resume(returning: data ?? Data())
|
connection.receive(minimumIncompleteLength: 1, maximumLength: 64 * 1024) { data, _, isComplete, error in
|
||||||
|
if let error {
|
||||||
|
readContinuation.resume(throwing: error)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if isComplete {
|
||||||
|
readContinuation.resume(throwing: SDLQUICError.dataStreamClosed)
|
||||||
|
} else {
|
||||||
|
readContinuation.resume(returning: data ?? Data())
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
} onCancel: {
|
||||||
|
readContinuation.resume(throwing: CancellationError())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -61,6 +61,7 @@ actor SDLSuperService {
|
|||||||
await superClient.stop()
|
await superClient.stop()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
await self.cleanup(superClient)
|
await self.cleanup(superClient)
|
||||||
} catch {
|
} catch {
|
||||||
await self.cleanup(superClient)
|
await self.cleanup(superClient)
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user