add quic client

This commit is contained in:
anlicheng 2026-02-14 23:06:51 +08:00
parent 57edebe42c
commit 78dc345d8b
3 changed files with 270 additions and 193 deletions

View File

@ -46,6 +46,9 @@ actor SDLContextActor {
private var dnsClient: SDLDNSClient? private var dnsClient: SDLDNSClient?
private var dnsWorker: Task<Void, Never>? private var dnsWorker: Task<Void, Never>?
private var quicClient: SDLQUICClient
private var quicWorker: Task<Void, Never>?
nonisolated private let puncherActor: SDLPuncherActor nonisolated private let puncherActor: SDLPuncherActor
// //
nonisolated private let proberActor: SDLNATProberActor nonisolated private let proberActor: SDLNATProberActor
@ -97,6 +100,13 @@ actor SDLContextActor {
public func start() { public func start() {
self.startMonitor() self.startMonitor()
self.loopChildWorkers.append(spawnLoop {
let quicClient = try await self.startQUICClient()
SDLLogger.shared.log("[SDLContext] quicClient running!!!!")
await quicClient.waitClose()
SDLLogger.shared.log("[SDLContext] quicClient closed!!!!")
})
self.loopChildWorkers.append(spawnLoop { self.loopChildWorkers.append(spawnLoop {
let noticeClient = try self.startNoticeClient() let noticeClient = try self.startNoticeClient()
SDLLogger.shared.log("[SDLContext] noticeClient running!!!!") SDLLogger.shared.log("[SDLContext] noticeClient running!!!!")
@ -119,6 +129,25 @@ actor SDLContextActor {
}) })
} }
private func startQUICClient() async throws -> SDLQUICClient {
// monitor
let quicClient = SDLQUICClient(host: "127.0.0.1", port: 1365)
quicClient.start()
// quic
try await quicClient.waitReady()
SDLLogger.shared.log("[SDLContext] start quic client ready")
self.quicWorker = Task.detached {
for await message in quicClient.receiveStream(maxLen: 86400) {
SDLLogger.shared.log("[SDLContext] quic client receive message: \(message)")
}
}
self.quicClient = quicClient
return quicClient
}
private func startNoticeClient() throws -> SDLNoticeClient { private func startNoticeClient() throws -> SDLNoticeClient {
// noticeClient // noticeClient
let noticeClient = try SDLNoticeClient(noticePort: self.config.noticePort, logger: SDLLogger.shared) let noticeClient = try SDLNoticeClient(noticePort: self.config.noticePort, logger: SDLLogger.shared)

View File

@ -0,0 +1,241 @@
//
// SDLQuicClient.swift
// Tun
//
// Created by on 2026/2/13.
//
import Foundation
import NIOCore
import Network
// 便
enum SDLQUICError: Error {
case connectionFailed(Error)
case connectionCancelled
case timeout
case decodeError(String)
}
final class SDLQUICClient {
private let transport: SDLQUICTransport
private let allocator = ByteBufferAllocator()
private let queue = DispatchQueue(label: "com.sdl.QUICClient.queue") // 线
private var closeCont: CheckedContinuation<Void, Never>?
private var readyCont: CheckedContinuation<Void, Error>?
init(host: String, port: UInt16) {
self.transport = SDLQUICTransport(host: host, port: port)
}
func start() {
self.transport.start(queue: self.queue) { event in
switch event {
case .ready:
self.readyCont?.resume()
self.readyCont = nil
case .failed(let error):
self.closeCont?.resume()
self.closeCont = nil
case .cancelled:
self.closeCont?.resume()
self.closeCont = nil
}
}
}
func receiveStream(maxLen: Int) -> AsyncCompactMapSequence<AsyncStream<Data>, SDLQUICInboundMessage> {
return transport.receiveMessageStream(maxLen: maxLen).compactMap { data in
var buf = self.allocator.buffer(bytes: data)
return try? QUICCodec.decode(buffer: &buf)
}
}
func send(data: Data) {
transport.send(data)
}
func waitReady() async throws {
return try await withCheckedThrowingContinuation { cont in
self.readyCont = cont
}
}
func waitClose() async {
return await withCheckedContinuation { cont in
self.closeCont = cont
}
}
deinit {
self.readyCont?.resume(throwing: SDLQUICError.connectionCancelled)
self.readyCont = nil
self.closeCont?.resume()
self.closeCont = nil
self.transport.stop()
}
}
final class SDLQUICTransport {
enum Event {
case ready
case failed(Error)
case cancelled
}
private let connection: NWConnection
init(host: String, port: UInt16) {
let params = NWParameters(quic: .init())
self.connection = NWConnection(host: .init(host), port: .init(rawValue: port)!, using: params)
}
func start(queue: DispatchQueue, onEvent: @escaping (Event) -> Void) {
connection.stateUpdateHandler = { state in
switch state {
case .ready: onEvent(.ready)
case .failed(let e): onEvent(.failed(e))
case .cancelled: onEvent(.cancelled)
default: break
}
}
connection.start(queue: queue)
}
func receiveMessageStream(maxLen: Int) -> AsyncStream<Data> {
let connection = self.connection
return AsyncStream { continuation in
var buffer = Data()
func tryParse() {
while true {
//
guard buffer.count >= 2 else {
return
}
let len0 = UInt16(bigEndian: buffer.withUnsafeBytes { $0.load(as: UInt16.self) })
let len = Int(len0)
//
guard buffer.count >= 2 + len else {
return
}
// body
let body = buffer.subdata(in: 2 ..< 2 + len)
continuation.yield(body)
//
buffer.removeSubrange(0 ..< 2 + len)
}
}
func loopReceive() {
connection.receive(minimumIncompleteLength: 1, maximumLength: maxLen) { data, _, _, error in
if let data, !data.isEmpty {
buffer.append(data)
tryParse()
}
if error == nil {
loopReceive()
} else {
continuation.finish()
}
}
}
loopReceive()
}
}
func send(_ data: Data) {
var len = UInt16(data.count).bigEndian
var packet = Data(Data(bytes: &len, count: 2))
packet.append(data)
connection.send(content: packet, completion: .contentProcessed { _ in })
}
func stop() {
connection.cancel()
}
}
extension SDLQUICClient {
struct QUICCodec {
// --MARK:
public static func decode(buffer: inout ByteBuffer) throws -> SDLQUICInboundMessage? {
guard let type = buffer.readInteger(as: UInt8.self),
let packetType = SDLPacketType(rawValue: type) else {
return nil
}
switch packetType {
case .registerSuperAck:
guard let bytes = buffer.readBytes(length: buffer.readableBytes),
let registerSuperAck = try? SDLRegisterSuperAck(serializedBytes: bytes) else {
return nil
}
return .registerSuperAck(registerSuperAck)
case .registerSuperNak:
guard let bytes = buffer.readBytes(length: buffer.readableBytes),
let registerSuperNak = try? SDLRegisterSuperNak(serializedBytes: bytes) else {
return nil
}
return .registerSuperNak(registerSuperNak)
case .peerInfo:
guard let bytes = buffer.readBytes(length: buffer.readableBytes),
let peerInfo = try? SDLPeerInfo(serializedBytes: bytes) else {
return nil
}
return .peerInfo(peerInfo)
case .policyResponse:
guard let bytes = buffer.readBytes(length: buffer.readableBytes),
let policyResponse = try? SDLPolicyResponse(serializedBytes: bytes) else {
return nil
}
return .policyReponse(policyResponse)
case .event:
guard let eventVal = buffer.readInteger(as: UInt8.self),
let event = SDLEventType(rawValue: eventVal),
let bytes = buffer.readBytes(length: buffer.readableBytes) else {
SDLLogger.shared.log("[SDLUDPHole] decode error 15")
return nil
}
switch event {
case .natChanged:
guard let natChangedEvent = try? SDLNatChangedEvent(serializedBytes: bytes) else {
SDLLogger.shared.log("[SDLUDPHole] decode error 16")
return nil
}
return .event(.natChanged(natChangedEvent))
case .sendRegister:
guard let sendRegisterEvent = try? SDLSendRegisterEvent(serializedBytes: bytes) else {
SDLLogger.shared.log("[SDLUDPHole] decode error 17")
return nil
}
return .event(.sendRegister(sendRegisterEvent))
case .networkShutdown:
guard let networkShutdownEvent = try? SDLNetworkShutdownEvent(serializedBytes: bytes) else {
SDLLogger.shared.log("[SDLUDPHole] decode error 18")
return nil
}
return .event(.networkShutdown(networkShutdownEvent))
}
default:
SDLLogger.shared.log("SDLUDPHole decode miss type: \(type)")
return nil
}
}
}
}

View File

@ -1,193 +0,0 @@
//
// SDLQuicClient.swift
// Tun
//
// Created by on 2026/2/13.
//
import Foundation
import NIOCore
import Network
actor SDLQUICClient {
private let transport: SDLQUICTransport
nonisolated private let alloctor = ByteBufferAllocator()
init(host: String, port: UInt16) {
self.transport = SDLQUICTransport(host: host, port: port)
}
func start() async -> SDLQUICTransport.Event {
return await withCheckedContinuation { cont in
transport.start { event in
cont.resume(returning: event)
}
}
}
func receiveStream(maxLen: Int) -> AsyncCompactMapSequence<AsyncStream<Data>, SDLQUICInboundMessage> {
return transport.receiveMessageStream(maxLen: maxLen).compactMap { data in
var buf = self.alloctor.buffer(bytes: data)
return try? Self.decode(buffer: &buf)
}
}
func send(data: Data) {
transport.send(data)
}
// --MARK:
private static func decode(buffer: inout ByteBuffer) throws -> SDLQUICInboundMessage? {
guard let type = buffer.readInteger(as: UInt8.self),
let packetType = SDLPacketType(rawValue: type) else {
return nil
}
switch packetType {
case .registerSuperAck:
guard let bytes = buffer.readBytes(length: buffer.readableBytes),
let registerSuperAck = try? SDLRegisterSuperAck(serializedBytes: bytes) else {
return nil
}
return .registerSuperAck(registerSuperAck)
case .registerSuperNak:
guard let bytes = buffer.readBytes(length: buffer.readableBytes),
let registerSuperNak = try? SDLRegisterSuperNak(serializedBytes: bytes) else {
return nil
}
return .registerSuperNak(registerSuperNak)
case .peerInfo:
guard let bytes = buffer.readBytes(length: buffer.readableBytes),
let peerInfo = try? SDLPeerInfo(serializedBytes: bytes) else {
return nil
}
return .peerInfo(peerInfo)
case .policyResponse:
guard let bytes = buffer.readBytes(length: buffer.readableBytes),
let policyResponse = try? SDLPolicyResponse(serializedBytes: bytes) else {
return nil
}
return .policyReponse(policyResponse)
case .event:
guard let eventVal = buffer.readInteger(as: UInt8.self),
let event = SDLEventType(rawValue: eventVal),
let bytes = buffer.readBytes(length: buffer.readableBytes) else {
SDLLogger.shared.log("[SDLUDPHole] decode error 15")
return nil
}
switch event {
case .natChanged:
guard let natChangedEvent = try? SDLNatChangedEvent(serializedBytes: bytes) else {
SDLLogger.shared.log("[SDLUDPHole] decode error 16")
return nil
}
return .event(.natChanged(natChangedEvent))
case .sendRegister:
guard let sendRegisterEvent = try? SDLSendRegisterEvent(serializedBytes: bytes) else {
SDLLogger.shared.log("[SDLUDPHole] decode error 17")
return nil
}
return .event(.sendRegister(sendRegisterEvent))
case .networkShutdown:
guard let networkShutdownEvent = try? SDLNetworkShutdownEvent(serializedBytes: bytes) else {
SDLLogger.shared.log("[SDLUDPHole] decode error 18")
return nil
}
return .event(.networkShutdown(networkShutdownEvent))
}
default:
SDLLogger.shared.log("SDLUDPHole decode miss type: \(type)")
return nil
}
}
}
final class SDLQUICTransport {
enum Event {
case ready
case failed(Error)
case cancelled
}
private let connection: NWConnection
init(host: String, port: UInt16) {
let params = NWParameters(quic: .init())
self.connection = NWConnection(host: .init(host), port: .init(rawValue: port)!, using: params)
}
func start(onEvent: @escaping (Event) -> Void) {
connection.stateUpdateHandler = { state in
switch state {
case .ready: onEvent(.ready)
case .failed(let e): onEvent(.failed(e))
case .cancelled: onEvent(.cancelled)
default: break
}
}
connection.start(queue: .global())
}
func receiveMessageStream(maxLen: Int) -> AsyncStream<Data> {
let connection = self.connection
return AsyncStream { continuation in
var buffer = Data()
func tryParse() {
while true {
//
guard buffer.count >= 2 else {
return
}
let len0 = UInt16(bigEndian: buffer.withUnsafeBytes { $0.load(as: UInt16.self) })
let len = Int(len0)
//
guard buffer.count >= 2 + len else {
return
}
// body
let body = buffer.subdata(in: 2 ..< 2 + len)
continuation.yield(body)
//
buffer.removeSubrange(0 ..< 2 + len)
}
}
func loopReceive() {
connection.receive(minimumIncompleteLength: 1, maximumLength: maxLen) { data, _, _, error in
if let data, !data.isEmpty {
buffer.append(data)
tryParse()
}
if error == nil {
loopReceive()
} else {
continuation.finish()
}
}
}
loopReceive()
}
}
func send(_ data: Data) {
var len = UInt16(data.count).bigEndian
var packet = Data(Data(bytes: &len, count: 2))
packet.append(data)
connection.send(content: packet, completion: .contentProcessed { _ in })
}
func stop() {
connection.cancel()
}
}