This commit is contained in:
anlicheng 2026-05-28 14:15:26 +08:00
parent 707bb82ea0
commit 8aeb543b17
2 changed files with 70 additions and 92 deletions

View File

@ -16,33 +16,36 @@ actor DNSCloudClient {
case invalidData case invalidData
} }
private enum State {
case idle
case starting
case running
case stopped
}
private var state: State = .idle
private let queue = DispatchQueue(label: "com.sdl.DNSCloudClient.queue") private let queue = DispatchQueue(label: "com.sdl.DNSCloudClient.queue")
private let connection: NWConnection
private var connection: NWConnection?
private let dnsServerAddress: NWEndpoint
// DNS // DNS
nonisolated let packetFlow: AsyncThrowingStream<Data, Error> nonisolated let packetFlow: AsyncThrowingStream<Data, Error>
private let packetContinuation: AsyncThrowingStream<Data, Error>.Continuation private let packetContinuation: AsyncThrowingStream<Data, Error>.Continuation
private var isPacketContinuationFinished: Bool = false
private let readySignal = AsyncOneShot<Void>() private let readySignal = AsyncOneShot<Void>()
private var isStopped: Bool = false
private var isPacketContinuationFinished: Bool = false
/// - Parameter serverIP: sn-server IP ( "8.8.8.8") /// - Parameter serverIP: sn-server IP ( "8.8.8.8")
/// - Parameter port: ( 53) /// - Parameter port: ( 53)
init(serverIP: String, port: UInt16) { init(serverIP: String, port: UInt16) {
self.dnsServerAddress = .hostPort(host: Self.makeEndpointHost(address: serverIP), port: NWEndpoint.Port(integerLiteral: port)) let dnsServerAddress = NWEndpoint.hostPort(host: Self.makeEndpointHost(address: serverIP), port: NWEndpoint.Port(integerLiteral: port))
let packetPair = AsyncThrowingStream.makeStream(of: Data.self) let packetPair = AsyncThrowingStream.makeStream(of: Data.self)
self.packetFlow = packetPair.stream self.packetFlow = packetPair.stream
self.packetContinuation = packetPair.continuation self.packetContinuation = packetPair.continuation
// 1.
let parameters = NWParameters.udp
// TUN NE TUN .other
parameters.prohibitedInterfaceTypes = [.other]
// 2. pathSelectionOptions
parameters.multipathServiceType = .handover
// 2.
self.connection = NWConnection(to: dnsServerAddress, using: parameters)
} }
private static func makeEndpointHost(address ip: String) -> NWEndpoint.Host { private static func makeEndpointHost(address ip: String) -> NWEndpoint.Host {
@ -58,46 +61,29 @@ actor DNSCloudClient {
} }
func run() async throws { func run() async throws {
guard self.state == .idle else { self.connection.stateUpdateHandler = { [weak self] state in
return
}
self.state = .starting
// 1.
let parameters = NWParameters.udp
// TUN NE TUN .other
parameters.prohibitedInterfaceTypes = [.other]
// 2. pathSelectionOptions
parameters.multipathServiceType = .handover
// 2.
let connection = NWConnection(to: self.dnsServerAddress, using: parameters)
connection.stateUpdateHandler = { [weak self] state in
Task { Task {
await self?.handleConnectionStateUpdate(state, for: connection) await self?.handleConnectionStateUpdate(state)
} }
} }
self.connection = connection self.connection.start(queue: self.queue)
//
connection.start(queue: self.queue)
try await withTaskCancellationHandler { try await withTaskCancellationHandler {
try await self.readySignal.wait() try await self.readySignal.wait()
while true { while true {
try Task.checkCancellation() try Task.checkCancellation()
let data = try await self.readOnce() let data = try await self.readOnce()
self.packetContinuation.yield(data) self.packetContinuation.yield(data)
} }
} onCancel: { } onCancel: {
connection.cancel() self.connection.cancel()
} }
} }
/// DNS TUN IP /// DNS TUN IP
func forward(ipPacketData: Data) { func forward(ipPacketData: Data) {
guard self.state == .running, guard connection.state == .ready else {
let connection = self.connection, connection.state == .ready else {
return return
} }
@ -111,30 +97,25 @@ actor DNSCloudClient {
} }
func stop() async { func stop() async {
guard self.state != .stopped else { guard !self.isStopped else {
return return
} }
self.isStopped = true
self.state = .stopped self.connection.cancel()
let connection = self.connection
self.connection = nil
connection?.cancel()
await self.readySignal.fail(DNSCloudError.cancelled) await self.readySignal.fail(DNSCloudError.cancelled)
self.finishPacketContinuationIfNeed(throwing: nil) self.finishPacketContinuationIfNeed(throwing: nil)
SDLLogger.log("[SDLCloudClient] stopped") SDLLogger.log("[SDLCloudClient] stopped")
} }
private func handleConnectionStateUpdate(_ state: NWConnection.State, for connection: NWConnection) async { private func handleConnectionStateUpdate(_ state: NWConnection.State) async {
switch state { switch state {
case .ready: case .ready:
guard self.state != .stopped, self.isCurrentConnection(connection) else { guard !self.isStopped else {
return return
} }
self.state = .running
SDLLogger.log("[DNSClient] Connection ready", for: .debug) SDLLogger.log("[DNSClient] Connection ready", for: .debug)
await self.readySignal.succeed(()) await self.readySignal.succeed(())
case .failed(let error): case .failed(let error):
@ -148,20 +129,13 @@ actor DNSCloudClient {
} }
} }
private func isCurrentConnection(_ connection: NWConnection) -> Bool {
guard let currentConnection = self.connection else {
return false
}
return currentConnection === connection
}
private func finishPacketContinuationIfNeed(throwing error: DNSCloudError?) { private func finishPacketContinuationIfNeed(throwing error: DNSCloudError?) {
guard !self.isPacketContinuationFinished else { guard !self.isPacketContinuationFinished else {
return return
} }
self.isPacketContinuationFinished = true self.isPacketContinuationFinished = true
if let error { if let error {
self.packetContinuation.finish(throwing: error) self.packetContinuation.finish(throwing: error)
} else { } else {
@ -170,21 +144,27 @@ actor DNSCloudClient {
} }
private func readOnce() async throws -> Data { private func readOnce() async throws -> Data {
guard let connection = self.connection, connection.state == .ready else { guard self.connection.state == .ready else {
throw DNSCloudError.cancelled throw DNSCloudError.cancelled
} }
return try await withCheckedThrowingContinuation { continuation in let readContinuation = OnceContinuation<Data, Error>()
connection.receiveMessage { content, _, _, error in return try await withTaskCancellationHandler {
try await withCheckedThrowingContinuation { cont in
readContinuation.set(cont)
self.connection.receiveMessage { content, _, _, error in
if let error { if let error {
continuation.resume(throwing: error) readContinuation.resume(throwing: error)
} else if let data = content, !data.isEmpty { } else if let data = content, !data.isEmpty {
continuation.resume(returning: data) readContinuation.resume(returning: data)
} else { } else {
continuation.resume(throwing: DNSCloudError.invalidData) readContinuation.resume(throwing: DNSCloudError.invalidData)
} }
} }
} }
} onCancel: {
readContinuation.resume(throwing: CancellationError())
}
} }
deinit { deinit {

View File

@ -20,35 +20,11 @@ actor DNSCloudService {
let client = DNSCloudClient(serverIP: self.serverIP, port: 15353) let client = DNSCloudClient(serverIP: self.serverIP, port: 15353)
self.currentClient = client self.currentClient = client
defer {
self.clearCurrent(client, generation: generation)
}
do { do {
try await self.run(client: client)
self.clearCurrent(client, generation: generation)
await client.stop()
} catch is CancellationError {
self.clearCurrent(client, generation: generation)
await client.stop()
throw CancellationError()
} catch {
self.clearCurrent(client, generation: generation)
await client.stop()
throw error
}
}
func stop() async {
self.generation &+= 1
let client = self.currentClient
self.currentClient = nil
await client?.stop()
}
func forward(ipPacketData: Data) async {
await self.currentClient?.forward(ipPacketData: ipPacketData)
}
private func run(client: DNSCloudClient) async throws {
let onEvent = self.onEvent let onEvent = self.onEvent
try await withThrowingTaskGroup(of: Void.self) { group in try await withThrowingTaskGroup(of: Void.self) { group in
@ -69,6 +45,28 @@ actor DNSCloudService {
_ = try await group.next() _ = try await group.next()
} }
await client.stop()
} catch is CancellationError {
await client.stop()
throw CancellationError()
} catch {
await client.stop()
throw error
}
}
func stop() async {
self.generation &+= 1
let client = self.currentClient
self.currentClient = nil
await client?.stop()
}
func forward(ipPacketData: Data) async {
await self.currentClient?.forward(ipPacketData: ipPacketData)
} }
private func nextGeneration() -> UInt64 { private func nextGeneration() -> UInt64 {