Package.resolved

@@ -1,5 +1,5 @@
 {
-  "originHash" : "597c0437584251872938aab4e6fd3cc1fc28e6da31c045adee36661e9c049c5e",
+  "originHash" : "c6ae4911d55046c288e0ecdc9de28a95dcb591d3daaa03914664a3d2e069977a",
   "pins" : [
     {
       "identity" : "swift-atomics",
@@ -24,8 +24,8 @@
       "kind" : "remoteSourceControl",
       "location" : "https://github.com/apple/swift-nio.git",
       "state" : {
-        "revision" : "a5fea865badcb1c993c85b0f0e8d05a4bd2270fb",
+        "revision" : "ad6b5f17270a7008f60d35ec5378e6144a575162",
-        "version" : "2.85.0"
+        "version" : "2.84.0"
       }
     },
     {

Package.swift

@@ -16,7 +16,7 @@ let package = Package(
             targets: ["Punchnet"]),
     ],
     dependencies: [
-        .package(url: "https://github.com/apple/swift-nio.git", exact: "2.85.0"),
+        .package(url: "https://github.com/apple/swift-nio.git", from: "2.84.0"),
         .package(url: "https://github.com/apple/swift-protobuf.git", from: "1.26.0")
         
     ],
@@ -28,7 +28,6 @@ let package = Package(
             dependencies: [
                 .product(name: "NIOCore", package: "swift-nio"),
                 .product(name: "NIOPosix", package: "swift-nio"),
-                .product(name: "NIOFoundationCompat", package: "swift-nio"),
                 .product(name: "SwiftProtobuf", package: "swift-protobuf")
             ]
         ),

Sources/Punchnet/ARPPacket.swift

@@ -31,15 +31,8 @@ struct ARPPacket {
     var targetMAC: Data
     var targetIP: UInt32
     
-    init(hardwareType: UInt16,
+    init(hardwareType: UInt16, protocolType: UInt16, hardwareSize: UInt8, protocolSize: UInt8, opcode: Opcode, 
-         protocolType: UInt16,
+         senderMAC: Data, senderIP: UInt32, targetMAC: Data, targetIP: UInt32) {
-         hardwareSize: UInt8,
-         protocolSize: UInt8,
-         opcode: Opcode,
-         senderMAC: Data,
-         senderIP: UInt32,
-         targetMAC: Data,
-         targetIP: UInt32) {
         
         self.hardwareType = hardwareType
         self.protocolType = protocolType

Sources/Punchnet/HolerManager.swift

@@ -0,0 +1,31 @@
+//
+//  HolerManager.swift
+//  sdlan
+//
+//  Created by 安礼成 on 2025/7/14.
+//
+import Foundation
+
+actor HolerManager {
+    private var holers: [Data:Task<(), Never>] = [:]
+    
+    func addHoler(dstMac: Data, creator: @escaping () -> Task<(), Never>) {
+        if let task = self.holers[dstMac] {
+            if task.isCancelled {
+                self.holers[dstMac] = creator()
+            }
+        } else {
+            self.holers[dstMac] = creator()
+        }
+    }
+    
+    func cleanup() {
+        for holer in holers.values {
+            holer.cancel()
+        }
+        self.holers.removeAll()
+    }
+    
+}
+
+

Sources/Punchnet/NetworkInterface.swift

@@ -0,0 +1,67 @@
+//
+//  NetworkInterface.swift
+//  Tun
+//
+//  Created by 安礼成 on 2024/1/19.
+//
+
+import Foundation
+
+public struct NetworkInterface {
+    public let name: String
+    public let ip: String
+    public let netmask: String
+}
+
+public struct NetworkInterfaceManager {
+    /**
+    获取网卡信息, (let name: String let ip: String let netmask: String)
+     */
+    public static func getInterfaces() -> [NetworkInterface] {
+        var interfaces: [NetworkInterface] = []
+        
+        // Get list of all interfaces on the local machine:
+        var ifaddr : UnsafeMutablePointer<ifaddrs>? = nil
+        if getifaddrs(&ifaddr) == 0 {
+            
+            // For each interface ...
+            var ptr = ifaddr
+            while( ptr != nil) {
+                
+                let flags = Int32(ptr!.pointee.ifa_flags)
+                var addr = ptr!.pointee.ifa_addr.pointee
+                
+                // Check for running IPv4, IPv6 interfaces. Skip the loopback interface.
+                if (flags & (IFF_UP|IFF_RUNNING|IFF_LOOPBACK)) == (IFF_UP|IFF_RUNNING) {
+                    if addr.sa_family == UInt8(AF_INET) || addr.sa_family == UInt8(AF_INET6) {
+                        
+                        var mask = ptr!.pointee.ifa_netmask.pointee
+                        
+                        // Convert interface address to a human readable string:
+                        let zero  = CChar(0)
+                        var hostname = [CChar](repeating: zero, count: Int(NI_MAXHOST))
+                        var netmask =  [CChar](repeating: zero, count: Int(NI_MAXHOST))
+                        if (getnameinfo(&addr, socklen_t(addr.sa_len), &hostname, socklen_t(hostname.count),
+                                        nil, socklen_t(0), NI_NUMERICHOST) == 0) {
+                            let address = String(cString: hostname)
+
+                            let name = ptr!.pointee.ifa_name!
+                            let ifname = String(cString: name)
+                            
+                            if (getnameinfo(&mask, socklen_t(mask.sa_len), &netmask, socklen_t(netmask.count), nil, socklen_t(0), NI_NUMERICHOST) == 0) {
+                                let netmaskIP = String(cString: netmask)
+                                
+                                interfaces.append(NetworkInterface(name: ifname, ip: address, netmask: netmaskIP))
+                            }
+                        }
+                    }
+                }
+                ptr = ptr!.pointee.ifa_next
+            }
+            freeifaddrs(ifaddr)
+        }
+        
+        return interfaces
+    }
+    
+}

Sources/Punchnet/SDLContext.swift

@@ -48,7 +48,10 @@ public class SDLContext: @unchecked Sendable {
     
     // 依赖的变量
     var udpHole: SDLUDPHole?
+    private var udpCancel: AnyCancellable?
+    
     var superClient: SDLSuperClient?
+    private var superCancel: AnyCancellable?
 
     // 数据包读取任务
     private var readTask: Task<(), Never>?
@@ -56,6 +59,7 @@ public class SDLContext: @unchecked Sendable {
     let provider: NEPacketTunnelProvider
     
     private var sessionManager: SessionManager
+    private var holerManager: HolerManager
     private var arpServer: ArpServer
     
     // 记录最后发送的stunRequest的cookie
@@ -69,23 +73,12 @@ public class SDLContext: @unchecked Sendable {
     private var monitorCancel: AnyCancellable?
     
     // 内部socket通讯
-    private var noticeClient: SDLNoticeClient?
+    private var noticeClient: SDLNoticeClient
     
     // 流量统计
     private var flowTracer = SDLFlowTracerActor()
     private var flowTracerCancel: AnyCancellable?
     
-    // 处理holer
-    private var holerPublishers: [Data:PassthroughSubject<RegisterRequest, Never>] = [:]
-    private var bag = Set<AnyCancellable>()
-    private var locker = NSLock()
-    
-    struct RegisterRequest {
-        let srcMac: Data
-        let dstMac: Data
-        let networkId: UInt32
-    }
-    
     public init(provider: NEPacketTunnelProvider, config: SDLConfiguration, rsaCipher: RSACipher, aesCipher: AESCipher) {
         self.config = config
         self.rsaCipher = rsaCipher
@@ -98,137 +91,59 @@ public class SDLContext: @unchecked Sendable {
         
         self.provider = provider
         self.sessionManager = SessionManager()
+        self.holerManager = HolerManager()
         self.arpServer = ArpServer(known_macs: [:])
+        self.noticeClient = SDLNoticeClient()
     }
     
     public func start() async throws {
-        self.noticeClient = try await SDLNoticeClient()
+        try await self.startSuperClient()
+        try await self.startUDPHole()
+        self.noticeClient.start()
         
-        try await withThrowingTaskGroup(of: Void.self) { group in
+        // 启动网络监控
-            group.addTask {
+        self.monitorCancel = self.monitor.eventFlow.sink { event in
-                while !Task.isCancelled {
+            switch event {
-                    do {
+            case .changed:
-                        try await self.startUDPHole()
+                // 需要重新探测网络的nat类型
-                    } catch let err {
+                Task {
-                        NSLog("udp Hole get err: \(err)")
+                    self.natType = await SDLNatProber.getNatType(udpHole: self.udpHole, config: self.config)
-                    }
+                    NSLog("didNetworkPathChanged, nat type is: \(self.natType)")
                 }
+            case .unreachable:
+                NSLog("didNetworkPathUnreachable")
             }
-            
-            group.addTask {
-                while !Task.isCancelled {
-                    do {
-                        try await self.startSuperClient()
-                    } catch let err {
-                        NSLog("[SDLContext] SuperClient get error: \(err), will restart")
-                        await self.arpServer.clear()
-                        try? await Task.sleep(for: .seconds(2))
-                    }
-                }
-            }
-            
-            group.addTask {
-                try await self.startMonitor()
-            }
-            
-            group.addTask {
-                while !Task.isCancelled {
-                    do {
-                        try await self.noticeClient?.start()
-                    } catch let err {
-                        NSLog("noticeClient get err: \(err)")
-                    }
-                }
-            }
-            
-            try await group.waitForAll()
         }
+        self.monitor.start()
     }
     
     public func stop() async {
+        self.superCancel?.cancel()
         self.superClient = nil
+        
+        self.udpCancel?.cancel()
         self.udpHole = nil
         
         self.readTask?.cancel()
     }
     
-    private func startUDPHole() async throws {
-        self.udpHole = try await SDLUDPHole()
-        
-        try await withThrowingTaskGroup(of: Void.self) { group in
-            group.addTask {
-                try await self.udpHole?.start()
-            }
-            
-            group.addTask {
-                while !Task.isCancelled {
-                    try await Task.sleep(nanoseconds: 5 * 1_000_000_000)
-                    self.lastCookie = await self.udpHole?.stunRequest(context: self)
-                }
-            }
-            
-            group.addTask {
-                if let eventFlow = self.udpHole?.eventFlow {
-                    for try await event in eventFlow {
-                        try await self.handleUDPEvent(event: event)
-                    }
-                }
-            }
-            try await group.waitForAll()
-        }
-    }
-    
     private func startSuperClient() async throws {
-        self.superClient = try await SDLSuperClient(host: self.config.superHost, port: self.config.superPort)
+        self.superClient = SDLSuperClient(host: config.superHost, port: config.superPort)
-        
+        // 建立super的绑定关系
-        try await withThrowingTaskGroup(of: Void.self) { group in
+        self.superCancel?.cancel()
-            group.addTask {
+        self.superCancel = self.superClient?.eventFlow.sink { event in
-                try await self.superClient?.start()
+            Task {
+                await self.handleSuperEvent(event: event)
             }
-            
-            group.addTask {
-                if let eventFlow = self.superClient?.eventFlow {
-                    for try await event in eventFlow {
-                        try await self.handleSuperEvent(event: event)
-                    }
-                }
-            }
-            try await group.waitForAll()
         }
+        try await self.superClient?.start()
     }
     
-    private func startMonitor() async throws {
+    private func handleSuperEvent(event: SDLSuperClient.SuperEvent) async {
-        try await withThrowingTaskGroup(of: Void.self) { group in
-            group.addTask {
-                try await self.noticeClient?.start()
-            }
-            
-            group.addTask {
-                // 启动网络监控
-                self.monitorCancel = self.monitor.eventFlow.sink { event in
-                    switch event {
-                    case .changed:
-                        // 需要重新探测网络的nat类型
-                        Task {
-                            self.natType = await SDLNatProber.getNatType(udpHole: self.udpHole, config: self.config)
-                            NSLog("didNetworkPathChanged, nat type is: \(self.natType)")
-                        }
-                    case .unreachable:
-                        NSLog("didNetworkPathUnreachable")
-                    }
-                }
-                self.monitor.start()
-            }
-            
-            try await group.waitForAll()
-        }
-    }
-    
-    private func handleSuperEvent(event: SDLSuperClient.SuperEvent) async throws {
         switch event {
         case .ready:
             NSLog("[SDLContext] get registerSuper, mac address: \(Self.formatMacAddress(mac: self.devAddr.mac))")
-            guard let message = try await self.superClient?.registerSuper(context: self).get() else {
+            guard let message = await self.superClient?.registerSuper(context: self) else {
                 return
             }
             
@@ -243,7 +158,7 @@ public class SDLContext: @unchecked Sendable {
                 
                 if upgradeType == .force {
                     let forceUpgrade = NoticeMessage.UpgradeMessage(prompt: registerSuperAck.upgradePrompt, address: registerSuperAck.upgradeAddress)
-                    await self.noticeClient?.send(data: forceUpgrade.binaryData)
+                    self.noticeClient.send(data: forceUpgrade.binaryData)
                     exit(-1)
                 }
                 
@@ -253,7 +168,7 @@ public class SDLContext: @unchecked Sendable {
                 
                 if upgradeType == .normal {
                     let normalUpgrade = NoticeMessage.UpgradeMessage(prompt: registerSuperAck.upgradePrompt, address: registerSuperAck.upgradeAddress)
-                    await self.noticeClient?.send(data: normalUpgrade.binaryData)
+                    self.noticeClient.send(data: normalUpgrade.binaryData)
                 }
                 
             case .registerSuperNak(let nakPacket):
@@ -265,17 +180,25 @@ public class SDLContext: @unchecked Sendable {
                 switch errorCode {
                 case .invalidToken, .nodeDisabled:
                     let alertNotice = NoticeMessage.AlertMessage(alert: errorMessage)
-                    await self.noticeClient?.send(data: alertNotice.binaryData)
+                    self.noticeClient.send(data: alertNotice.binaryData)
                     exit(-1)
                 case .noIpAddress, .networkFault, .internalFault:
                     let alertNotice = NoticeMessage.AlertMessage(alert: errorMessage)
-                    await self.noticeClient?.send(data: alertNotice.binaryData)
+                    self.noticeClient.send(data: alertNotice.binaryData)
                 }
                 NSLog("[SDLContext] Get a SuperNak message exit")
             default:
                 ()
             }
             
+        case .closed:
+            NSLog("[SDLContext] super client closed")
+            await self.arpServer.clear()
+            DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
+                Task {@MainActor in
+                    try await self.startSuperClient()
+                }
+            }
         case .event(let evt):
             switch evt {
             case .natChanged(let natChangedEvent):
@@ -287,12 +210,12 @@ public class SDLContext: @unchecked Sendable {
                 let address = SDLUtil.int32ToIp(sendRegisterEvent.natIp)
                 if let remoteAddress = try? SocketAddress.makeAddressResolvingHost(address, port: Int(sendRegisterEvent.natPort)) {
                     // 发送register包
-                    await self.udpHole?.sendRegister(remoteAddress: remoteAddress, networkId: self.devAddr.networkID, srcMac: self.devAddr.mac, dst_mac: sendRegisterEvent.dstMac)
+                    self.udpHole?.sendRegister(context: self, remoteAddress: remoteAddress, dst_mac: sendRegisterEvent.dstMac)
                 }
                 
             case .networkShutdown(let shutdownEvent):
                 let alertNotice = NoticeMessage.AlertMessage(alert: shutdownEvent.message)
-                await self.noticeClient?.send(data: alertNotice.binaryData)
+                self.noticeClient.send(data: alertNotice.binaryData)
                 exit(-1)
             }
         case .command(let packetId, let command):
@@ -310,19 +233,44 @@ public class SDLContext: @unchecked Sendable {
                 var commandAck = SDLCommandAck()
                 commandAck.status = true
                 
-                await self.superClient?.commandAck(packetId: packetId, ack: commandAck)
+                self.superClient?.commandAck(packetId: packetId, ack: commandAck)
             }
         }
         
     }
 
-    private func handleUDPEvent(event: SDLUDPHole.UDPEvent) async throws {
+    private func startUDPHole() async throws {
+        self.udpHole = SDLUDPHole()
+        
+        self.udpCancel?.cancel()
+        self.udpCancel = self.udpHole?.eventFlow.sink { event in
+            Task.detached {
+                await self.handleUDPEvent(event: event)
+            }
+        }
+        
+        try await self.udpHole?.start()
+    }
+    
+    private func handleUDPEvent(event: SDLUDPHole.UDPEvent) async {
         switch event {
         case .ready:
             // 获取当前网络的类型
-            //self.natType = await SDLNatProber.getNatType(udpHole: self.udpHole, config: self.config)
+            self.natType = await SDLNatProber.getNatType(udpHole: self.udpHole, config: self.config)
             SDLLogger.log("[SDLContext] nat type is: \(self.natType)", level: .debug)
             
+            let timer = Timer.publish(every: 5.0, on: .main, in: .common).autoconnect()
+            self.stunCancel = Just(Date()).merge(with: timer).sink { _ in
+                self.lastCookie = self.udpHole?.stunRequest(context: self)
+            }
+            
+        case .closed:
+            DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
+                Task {
+                    try await self.startUDPHole()
+                }
+            }
+            
         case .message(let remoteAddress, let message):
             switch message {
             case .register(let register):
@@ -330,7 +278,7 @@ public class SDLContext: @unchecked Sendable {
                 // 判断目标地址是否是tun的网卡地址, 并且是在同一个网络下
                 if register.dstMac == self.devAddr.mac && register.networkID == self.devAddr.networkID {
                     // 回复ack包
-                    await self.udpHole?.sendRegisterAck(context: self, remoteAddress: remoteAddress, dst_mac: register.srcMac)
+                    self.udpHole?.sendRegisterAck(context: self, remoteAddress: remoteAddress, dst_mac: register.srcMac)
                     // 这里需要建立到来源的会话, 在复杂网络下，通过super-node查询到的nat地址不一定靠谱，需要通过udp包的来源地址作为nat地址
                     let session = Session(dstMac: register.srcMac, natAddress: remoteAddress)
                     await self.sessionManager.addSession(session: session)
@@ -412,18 +360,18 @@ public class SDLContext: @unchecked Sendable {
     }
     
     // 流量统计
-//    public func flowReportTask() {
+    public func flowReportTask() {
-//        Task {
+        Task {
-//            // 每分钟汇报一次
+            // 每分钟汇报一次
-//            self.flowTracerCancel = Timer.publish(every: 60.0, on: .main, in: .common).autoconnect()
+            self.flowTracerCancel = Timer.publish(every: 60.0, on: .main, in: .common).autoconnect()
-//                .sink { _ in
+                .sink { _ in
-//                    Task {
+                    Task {
-//                        let (forwardNum, p2pNum, inboundNum) = await self.flowTracer.reset()
+                        let (forwardNum, p2pNum, inboundNum) = await self.flowTracer.reset()
-//                        await self.superClient?.flowReport(forwardNum: forwardNum, p2pNum: p2pNum, inboundNum: inboundNum)
+                        self.superClient?.flowReport(forwardNum: forwardNum, p2pNum: p2pNum, inboundNum: inboundNum)
-//                    }
+                    }
-//                }
+                }
-//        }
+        }
-//    }
+    }
     
     // 网络改变时需要重新配置网络信息
     private func didNetworkConfigChanged(devAddr: SDLDevAddr, dnsServers: [String]? = nil) async {
@@ -453,6 +401,8 @@ public class SDLContext: @unchecked Sendable {
         // 网卡配置设置必须成功
         do {
             try await self.provider.setTunnelNetworkSettings(networkSettings)
+            
+            await self.holerManager.cleanup()
             self.startReader()
             
             NSLog("[SDLContext] setTunnelNetworkSettings success, start read packet")
@@ -493,7 +443,7 @@ public class SDLContext: @unchecked Sendable {
                             else {
                                 // 构造arp请求
                                 let broadcastMac = Data([0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF])
-                                let arpReqeust = ARPPacket.arpRequest(senderIP: self.devAddr.netAddr, senderMAC: self.devAddr.mac, targetIP: dstIp)
+                                let arpReqeust: ARPPacket = ARPPacket.arpRequest(senderIP: self.devAddr.netAddr, senderMAC: self.devAddr.mac, targetIP: dstIp)
                                 await self.routeLayerPacket(dstMac: broadcastMac, type: .arp, data: arpReqeust.marshal())
                                 
                                 NSLog("[SDLContext] dstIp: \(dstIp) arp query not found")
@@ -517,63 +467,43 @@ public class SDLContext: @unchecked Sendable {
         // 通过session发送到对端
         if let session = await self.sessionManager.getSession(toAddress: dstMac) {
             NSLog("[SDLContext] send packet by session: \(session)")
-            await self.udpHole?.sendPacket(context: self, session: session, data: encodedPacket)
+            self.udpHole?.sendPacket(context: self, session: session, data: encodedPacket)
             
             await self.flowTracer.inc(num: data.count, type: .p2p)
         }
         else {
             // 通过super_node进行转发
-            await self.udpHole?.forwardPacket(context: self, dst_mac: dstMac, data: encodedPacket)
+            self.udpHole?.forwardPacket(context: self, dst_mac: dstMac, data: encodedPacket)
             // 流量统计
             await self.flowTracer.inc(num: data.count, type: .forward)
             
             // 尝试打洞
-            let registerRequest = RegisterRequest(srcMac: self.devAddr.mac, dstMac: dstMac, networkId: self.devAddr.networkID)
+            await self.holerManager.addHoler(dstMac: dstMac) {
-            self.submitRegisterRequest(request: registerRequest)
+                self.holerTask(dstMac: dstMac)
-        }
+            }
-    }
+        }
-    
+    }
-    private func submitRegisterRequest(request: RegisterRequest) {
+    
-        self.locker.lock()
+    func holerTask(dstMac: Data) -> Task<(), Never> {
-        defer {
+        return Task {
-            self.locker.unlock()
+            guard let message = try? await self.superClient?.queryInfo(context: self, dst_mac: dstMac) else {
-        }
+                return
-        
+            }
-        let dstMac = request.dstMac
+            
-        if let publisher = self.holerPublishers[dstMac] {
+            switch message.packet {
-            publisher.send(request)
+            case .empty:
-        } else {
+                SDLLogger.log("[SDLContext] hole query_info get empty: \(message)", level: .debug)
-            let publisher = PassthroughSubject<RegisterRequest, Never>()
+            case .peerInfo(let peerInfo):
-            publisher.throttle(for: .seconds(5), scheduler: DispatchQueue.global(), latest: true)
+                if let remoteAddress = peerInfo.v4Info.socketAddress() {
-                .sink { request in
+                    SDLLogger.log("[SDLContext] hole sock address: \(remoteAddress)", level: .warning)
-                    Task {
+                    // 发送register包
-                        await self.tryHole(request: request)
+                    self.udpHole?.sendRegister(context: self, remoteAddress: remoteAddress, dst_mac: dstMac)
-                    }
+                } else {
-                }
+                    SDLLogger.log("[SDLContext] hole sock address is invalid: \(peerInfo.v4Info)", level: .warning)
-                .store(in: &self.bag)
+                }
-            
+            default:
-            self.holerPublishers[dstMac] = publisher
+                SDLLogger.log("[SDLContext] hole query_info is packet: \(message)", level: .warning)
-        }
-    }
-    
-    private func tryHole(request: RegisterRequest) async {
-        guard let message = try? await self.superClient?.queryInfo(dst_mac: request.dstMac).get() else {
-            return
-        }
-        
-        switch message.packet {
-        case .empty:
-            SDLLogger.log("[SDLContext] hole query_info get empty: \(message)", level: .debug)
-        case .peerInfo(let peerInfo):
-            if let remoteAddress = peerInfo.v4Info.socketAddress() {
-                SDLLogger.log("[SDLContext] hole sock address: \(remoteAddress)", level: .warning)
-                // 发送register包
-                await self.udpHole?.sendRegister(remoteAddress: remoteAddress, networkId: request.networkId, srcMac: request.srcMac, dst_mac: request.dstMac)
-            } else {
-                SDLLogger.log("[SDLContext] hole sock address is invalid: \(peerInfo.v4Info)", level: .warning)
             }
-        default:
-            SDLLogger.log("[SDLContext] hole query_info is packet: \(message)", level: .warning)
         }
     }
     

Sources/Punchnet/SDLError.swift

@@ -1,11 +0,0 @@
-//
-//  SDLError.swift
-//  sdlan
-//
-//  Created by 安礼成 on 2025/8/2.
-//
-
-enum SDLError: Error {
-    case socketClosed
-    case socketError
-}

Sources/Punchnet/SDLMessage.swift

@@ -54,7 +54,7 @@ enum SDLUpgradeType: UInt32 {
 }
 
 // Id生成器
-struct SDLIdGenerator: Sendable {
+struct SDLIdGenerator {
     // 消息体id
     private var packetId: UInt32
     

Sources/Punchnet/SDLNatProber.swift

@@ -34,7 +34,7 @@ struct SDLNatProber {
         }
         
         // 网络没有在nat下
-        if await natAddress1 == udpHole.localAddress {
+        if natAddress1 == udpHole.localAddress {
             return .noNat
         }
         
@@ -68,7 +68,7 @@ struct SDLNatProber {
     }
     
     private static func getNatAddress(_ udpHole: SDLUDPHole, remoteAddress: SocketAddress, attr: SDLProbeAttr) async -> SocketAddress? {
-        let stunProbeReply = try? await udpHole.stunProbe(remoteAddress: remoteAddress, attr: attr, timeout: 5)
+        let stunProbeReply = await udpHole.stunProbe(remoteAddress: remoteAddress, attr: attr, timeout: 5)
         
         return stunProbeReply?.socketAddress()
     }

Sources/Punchnet/SDLNoticeClient.swift

@@ -15,65 +15,75 @@ import Foundation
 //
 
 import Foundation
-import NIOCore
+@preconcurrency import NIOCore
 import NIOPosix
 
 // 处理和sn-server服务器之间的通讯
-actor SDLNoticeClient {
+class SDLNoticeClient: ChannelInboundHandler, @unchecked Sendable {
+    public typealias InboundIn = AddressedEnvelope<ByteBuffer>
+    public typealias OutboundOut = AddressedEnvelope<ByteBuffer>
+    
+    var channel: Channel?
     private let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
-    private let asyncChannel: NIOAsyncChannel<AddressedEnvelope<ByteBuffer>, AddressedEnvelope<ByteBuffer>>
     private let remoteAddress: SocketAddress
-    private let (writeStream, writeContinuation) = AsyncStream.makeStream(of: Data.self, bufferingPolicy: .unbounded)
     
-    // 启动函数
+    init() {
-    init() async throws {
         self.remoteAddress = try! SocketAddress(ipAddress: "127.0.0.1", port: 50195)
-        
-        let bootstrap = DatagramBootstrap(group: self.group)
-            .channelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
-        
-        self.asyncChannel = try await bootstrap.bind(host: "0.0.0.0", port: 0)
-            .flatMapThrowing {channel in
-                return try NIOAsyncChannel(wrappingChannelSynchronously: channel, configuration: .init(
-                    inboundType: AddressedEnvelope<ByteBuffer>.self,
-                    outboundType: AddressedEnvelope<ByteBuffer>.self
-                ))
-            }
-            .get()
-        
-        SDLLogger.log("[SDLNoticeClient] started and listening on: \(self.asyncChannel.channel.localAddress!)", level: .debug)
     }
     
-    func start() async throws {
+    // 启动函数
-        try await self.asyncChannel.executeThenClose { inbound, outbound in
+    func start() {
-            try await withThrowingTaskGroup(of: Void.self) { group in
+        let bootstrap = DatagramBootstrap(group: self.group)
-                group.addTask {
+            .channelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
-                    try await self.asyncChannel.channel.closeFuture.get()
+            .channelInitializer { channel in
-                    NSLog("[UDPHole] channel closed")
+                // 接收缓冲区
-                    throw SDLError.socketClosed
+                channel.pipeline.addHandler(self)
-                }
-                
-                group.addTask {
-                    for try await message in self.writeStream {
-                        let buf = self.asyncChannel.channel.allocator.buffer(bytes: message)
-                        let envelope = AddressedEnvelope<ByteBuffer>(remoteAddress: self.remoteAddress, data: buf)
-                        
-                        try await outbound.write(envelope)
-                    }
-                }
-                
-                try await group.waitForAll()
             }
-        }
+        
+        self.channel = try! bootstrap.bind(host: "0.0.0.0", port: 0).wait()
+        SDLLogger.log("[SDLNoticeClient] started and listening on: \(self.channel?.localAddress!)", level: .debug)
+    }
+    
+    // -- MARK: ChannelInboundHandler Methods
+    
+    public func channelActive(context: ChannelHandlerContext) {
+        
+    }
+    
+    // 接收到的消息, 消息需要根据类型分流
+    public func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+        context.fireChannelRead(data)
+    }
+    
+    public func errorCaught(context: ChannelHandlerContext, error: Error) {
+        // As we are not really interested getting notified on success or failure we just pass nil as promise to
+        // reduce allocations.
+        context.close(promise: nil)
+        self.channel = nil
+    }
+    
+    public func channelInactive(context: ChannelHandlerContext) {
+        self.channel = nil
+        context.close(promise: nil)
     }
     
     // 处理写入逻辑
     func send(data: Data) {
-        self.writeContinuation.yield(data)
+        guard let channel = self.channel else {
+            return
+        }
+        
+        let remoteAddress = self.remoteAddress
+        let allocator = channel.allocator
+        
+        channel.eventLoop.execute { [allocator] in
+            let buffer = allocator.buffer(bytes: data)
+            let envelope = AddressedEnvelope<ByteBuffer>(remoteAddress: remoteAddress, data: buffer)
+            channel.writeAndFlush(self.wrapOutboundOut(envelope), promise: nil)
+        }
     }
     
     deinit {
         try? self.group.syncShutdownGracefully()
-        self.writeContinuation.finish()
     }
 }

Sources/Punchnet/SDLSuperClient.swift

@@ -8,107 +8,59 @@
 import Foundation
 import NIOCore
 import NIOPosix
+import Combine
 
 // --MARK: 和SuperNode的客户端
-actor SDLSuperClient {
+class SDLSuperClient: ChannelInboundHandler, @unchecked Sendable {
-    private let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+    public typealias InboundIn = ByteBuffer
-    private let asyncChannel: NIOAsyncChannel<ByteBuffer,ByteBuffer>
+    public typealias OutboundOut = ByteBuffer
-    private let (writeStream, writeContinuation) = AsyncStream.makeStream(of: TcpMessage.self, bufferingPolicy: .unbounded)
-    private var callbackPromises: [UInt32:EventLoopPromise<SDLSuperInboundMessage>] = [:]
     
-    public let (eventFlow, inboundContinuation) = AsyncStream.makeStream(of: SuperEvent.self, bufferingPolicy: .unbounded)
+    public typealias CallbackFun = (SDLSuperInboundMessage?) -> Void
+    
+    private let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+    private var channel: Channel?
     
     // id生成器
     var idGenerator = SDLIdGenerator(seed: 1)
+    private var callbackManager = SuperCallbackManager()
     
-    // 发送的消息格式
+    let host: String
-    struct TcpMessage {
+    let port: Int
-        let packetId: UInt32
+    
-        let type: SDLPacketType
+    private var pingCancel: AnyCancellable?
-        let data: Data
+    
-    }
+    public var eventFlow = PassthroughSubject<SuperEvent, Never>()
     
     // 定义事件类型
     enum SuperEvent {
         case ready
+        case closed
         case event(SDLEvent)
         case command(UInt32, SDLCommand)
     }
     
-    init(host: String, port: Int) async throws {
+    init(host: String, port: Int) {
+        self.host = host
+        self.port = port
+    }
+    
+    func start() async throws {
         let bootstrap = ClientBootstrap(group: self.group)
             .channelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
             .channelInitializer { channel in
                 return channel.pipeline.addHandlers([
-                    ByteToMessageHandler(FixedHeaderDecoder()),
+                    ByteToMessageHandler(FixedHeaderDelimiterCoder()),
-                    MessageToByteHandler(FixedHeaderEncoder())
+                    MessageToByteHandler(FixedHeaderDelimiterCoder()),
+                    self
                 ])
             }
         
-        self.asyncChannel = try await bootstrap.connect(host: host, port: port)
+        do {
-            .flatMapThrowing { channel in
+            NSLog("super client connect: \(self.host):\(self.port)")
-                return try NIOAsyncChannel(wrappingChannelSynchronously: channel, configuration: .init(
+            self.channel = try await bootstrap.connect(host: self.host, port: self.port).get()
-                    inboundType: ByteBuffer.self,
+        } catch let err {
-                    outboundType: ByteBuffer.self
+            NSLog("super client get error: \(err)")
-                ))
+            self.eventFlow.send(.closed)
-            }
-            .get()
-    }
-    
-    func start() async throws {
-        try await self.asyncChannel.executeThenClose { inbound, outbound in
-            self.inboundContinuation.yield(.ready)
-            
-            try await withThrowingTaskGroup(of: Void.self) { group in
-                group.addTask {
-                    try await self.asyncChannel.channel.closeFuture.get()
-                    NSLog("[SDLSuperClient] socket closed")
-                    self.writeContinuation.finish()
-                    self.inboundContinuation.finish()
-                    
-                    throw SDLError.socketClosed
-                }
-                
-                group.addTask {
-                    for try await var packet in inbound {
-                        if let message = SDLSuperClientDecoder.decode(buffer: &packet) {
-                            SDLLogger.log("[SDLSuperTransport] read message: \(message)", level: .warning)
-                            switch message.packet {
-                            case .event(let event):
-                                self.inboundContinuation.yield(.event(event))
-                            case .command(let command):
-                                self.inboundContinuation.yield(.command(message.msgId, command))
-                            default:
-                                await self.fireCallback(message: message)
-                            }
-                        }
-                    }
-                    NSLog("[SDLSuperClient] inbound closed")
-                }
-                
-                group.addTask {
-                    for try await message in self.writeStream {
-                        var buffer = self.asyncChannel.channel.allocator.buffer(capacity: message.data.count + 5)
-                        buffer.writeInteger(message.packetId, as: UInt32.self)
-                        buffer.writeBytes([message.type.rawValue])
-                        buffer.writeBytes(message.data)
-                        try await outbound.write(buffer)
-                    }
-                    NSLog("[SDLSuperClient] outbound closed")
-                }
-                
-                // --MARK: 心跳机制
-                group.addTask {
-                    while !Task.isCancelled {
-                        await self.ping()
-                        try await Task.sleep(nanoseconds: 5 * 1_000_000_000)
-                    }
-                    NSLog("[SDLSuperClient] heartbeat cancelled")
-                }
-                
-                try await group.waitForAll()
-                NSLog("[SDLSuperClient] group closed")
-            }
         }
     }
     
@@ -122,7 +74,15 @@ actor SDLSuperClient {
         self.send(type: .commandAck, packetId: packetId, data: data)
     }
     
-    func registerSuper(context ctx: SDLContext) throws -> EventLoopFuture<SDLSuperInboundMessage> {
+    func registerSuper(context ctx: SDLContext) async -> SDLSuperInboundMessage? {
+        return await withCheckedContinuation { c in
+            self.registerSuper(context: ctx) { message in
+                c.resume(returning: message)
+            }
+        }
+    }
+    
+    func registerSuper(context ctx: SDLContext, callback: @escaping CallbackFun) {
         var registerSuper = SDLRegisterSuper()
         registerSuper.version = UInt32(ctx.config.version)
         registerSuper.clientID = ctx.config.clientId
@@ -132,15 +92,23 @@ actor SDLSuperClient {
         
         let data = try! registerSuper.serializedData()
         
-        return self.write(type: .registerSuper, data: data)
+        self.write(type: .registerSuper, data: data, callback: callback)
+    }
+    
+    func queryInfo(context ctx: SDLContext, dst_mac: Data) async throws -> SDLSuperInboundMessage? {
+        return await withCheckedContinuation { c in
+            self.queryInfo(context: ctx, dst_mac: dst_mac) { message in
+                c.resume(returning: message)
+            }
+        }
     }
     
     // 查询目标服务器的相关信息
-    func queryInfo(dst_mac: Data) async throws -> EventLoopFuture<SDLSuperInboundMessage> {
+    func queryInfo(context ctx: SDLContext, dst_mac: Data, callback: @escaping CallbackFun) {
         var queryInfo = SDLQueryInfo()
         queryInfo.dstMac = dst_mac
         
-        return self.write(type: .queryInfo, data: try! queryInfo.serializedData())
+        self.write(type: .queryInfo, data: try! queryInfo.serializedData(), callback: callback)
     }
     
     func unregister(context ctx: SDLContext) throws {
@@ -160,41 +128,156 @@ actor SDLSuperClient {
         self.send(type: .flowTracer, packetId: 0, data: try! flow.serializedData())
     }
     
-    private func write(type: SDLPacketType, data: Data) -> EventLoopFuture<SDLSuperInboundMessage> {
+    // --MARK: ChannelInboundHandler
-        SDLLogger.log("[SDLSuperTransport] will write data: \(data)", level: .debug)
-        let packetId = idGenerator.nextId()
-        let promise = self.asyncChannel.channel.eventLoop.makePromise(of: SDLSuperInboundMessage.self)
-        self.callbackPromises[packetId] = promise
     
-        self.writeContinuation.yield(TcpMessage(packetId: packetId, type: type, data: data))
+    public func channelActive(context: ChannelHandlerContext) {
-        
+        self.startPingTicker()
-        return promise.futureResult
+        self.eventFlow.send(.ready)
     }
     
-    private func send(type: SDLPacketType, packetId: UInt32, data: Data) {
+    public func channelRead(context: ChannelHandlerContext, data: NIOAny) {
-        self.writeContinuation.yield(TcpMessage(packetId: packetId, type: type, data: data))
+        var buffer = self.unwrapInboundIn(data)
-    }
+        if let message = decode(buffer: &buffer) {
+            SDLLogger.log("[SDLSuperTransport] read message: \(message)", level: .warning)
             
-    // 处理回调函数
+            switch message.packet {
-    private func fireCallback(message: SDLSuperInboundMessage) {
+            case .event(let event):
-        if let promise = self.callbackPromises[message.msgId] {
+                self.eventFlow.send(.event(event))
-            self.asyncChannel.channel.eventLoop.execute {
+            case .command(let command):
-                promise.succeed(message)
+                self.eventFlow.send(.command(message.msgId, command))
+            default:
+                self.callbackManager.fireCallback(message: message)
             }
-            self.callbackPromises.removeValue(forKey: message.msgId)
         }
     }
     
+    public func errorCaught(context: ChannelHandlerContext, error: Error) {
+        SDLLogger.log("[SDLSuperTransport] error: \(error)", level: .warning)
+        self.channel = nil
+        self.eventFlow.send(.closed)
+        context.close(promise: nil)
+    }
+    
+    public func channelInactive(context: ChannelHandlerContext) {
+        SDLLogger.log("[SDLSuperTransport] channelInactive", level: .warning)
+        self.channel = nil
+        context.close(promise: nil)
+    }
+    
+    func write(type: SDLPacketType, data: Data, callback: @escaping CallbackFun) {
+        guard let channel = self.channel else {
+            return
+        }
+        
+        SDLLogger.log("[SDLSuperTransport] will write data: \(data)", level: .debug)
+        
+        let packetId = idGenerator.nextId()
+        self.callbackManager.addCallback(id: packetId, callback: callback)
+        
+        channel.eventLoop.execute {
+            var buffer = channel.allocator.buffer(capacity: data.count + 5)
+            buffer.writeInteger(packetId, as: UInt32.self)
+            buffer.writeBytes([type.rawValue])
+            buffer.writeBytes(data)
+            
+            channel.writeAndFlush(self.wrapOutboundOut(buffer), promise: nil)
+        }
+    }
+    
+    func send(type: SDLPacketType, packetId: UInt32, data: Data) {
+        guard let channel = self.channel else {
+            return
+        }
+        
+        channel.eventLoop.execute {
+            var buffer = channel.allocator.buffer(capacity: data.count + 5)
+            buffer.writeInteger(packetId, as: UInt32.self)
+            buffer.writeBytes([type.rawValue])
+            buffer.writeBytes(data)
+            
+            channel.writeAndFlush(self.wrapOutboundOut(buffer), promise: nil)
+        }
+    }
+    
+    // --MARK: 心跳机制
+    
+    private func startPingTicker() {
+        self.pingCancel = Timer.publish(every: 5.0, on: .main, in: .common).autoconnect()
+            .sink { _ in
+                // 保持和super-node的心跳机制
+                self.ping()
+            }
+    }
+    
     deinit {
+        self.pingCancel?.cancel()
         try! group.syncShutdownGracefully()
     }
     
 }
 
+/// 基于2字节固定长度的分包协议
+extension SDLSuperClient {
+    private final class FixedHeaderDelimiterCoder: ByteToMessageDecoder, MessageToByteEncoder {
+        typealias InboundIn = ByteBuffer
+        typealias InboundOut = ByteBuffer
+        
+        func decode(context: ChannelHandlerContext, buffer: inout ByteBuffer) throws -> DecodingState {
+            guard let len = buffer.getInteger(at: buffer.readerIndex, endianness: .big, as: UInt16.self) else {
+                return .needMoreData
+            }
+            
+            if buffer.readableBytes >= len + 2 {
+                buffer.moveReaderIndex(forwardBy: 2)
+                if let bytes = buffer.readBytes(length: Int(len)) {
+                    context.fireChannelRead(self.wrapInboundOut(ByteBuffer(bytes: bytes)))
+                }
+                return .continue
+            } else {
+                return .needMoreData
+            }
+        }
+        
+        func encode(data: ByteBuffer, out: inout ByteBuffer) throws {
+            let len = data.readableBytes
+            out.writeInteger(UInt16(len))
+            out.writeBytes(data.readableBytesView)
+        }
+    }
+}
+
+// 回调函数管理器
+extension SDLSuperClient {
+    
+    private struct SuperCallbackManager {
+        // 对应请求体和相应的关系
+        private var callbacks: [UInt32:CallbackFun] = [:]
+        
+        mutating func addCallback(id: UInt32, callback: @escaping CallbackFun) {
+            self.callbacks[id] = callback
+        }
+        
+        mutating func fireCallback(message: SDLSuperInboundMessage) {
+            if let callback = self.callbacks[message.msgId] {
+                callback(message)
+                self.callbacks.removeValue(forKey: message.msgId)
+            }
+        }
+        
+        mutating func fireAllCallbacks(message: SDLSuperInboundMessage) {
+            for (_, callback) in self.callbacks {
+                callback(nil)
+            }
+            self.callbacks.removeAll()
+        }
+    }
+    
+}
+
 // --MARK: 编解码器
-private struct SDLSuperClientDecoder {
+extension SDLSuperClient {
     // 消息格式为: <<MsgId:32, Type:8, Body/binary>>
-    static func decode(buffer: inout ByteBuffer) -> SDLSuperInboundMessage? {
+    func decode(buffer: inout ByteBuffer) -> SDLSuperInboundMessage? {
         guard let msgId = buffer.readInteger(as: UInt32.self),
               let type = buffer.readInteger(as: UInt8.self),
               let messageType = SDLPacketType(rawValue: type) else {
@@ -273,36 +356,13 @@ private struct SDLSuperClientDecoder {
             return nil
         }
     }
+    
 }
 
-private final class FixedHeaderEncoder: MessageToByteEncoder, @unchecked Sendable {
+extension ByteToMessageHandler: @unchecked @retroactive Sendable {
-    typealias InboundIn = ByteBuffer
-    typealias InboundOut = ByteBuffer
     
-    func encode(data: ByteBuffer, out: inout ByteBuffer) throws {
-        let len = data.readableBytes
-        out.writeInteger(UInt16(len))
-        out.writeBytes(data.readableBytesView)
-    }
 }
 
-private final class FixedHeaderDecoder: ByteToMessageDecoder, @unchecked Sendable {
+extension MessageToByteHandler: @unchecked @retroactive Sendable {
-    typealias InboundIn = ByteBuffer
-    typealias InboundOut = ByteBuffer
     
-    func decode(context: ChannelHandlerContext, buffer: inout ByteBuffer) throws -> DecodingState {
-        guard let len = buffer.getInteger(at: buffer.readerIndex, endianness: .big, as: UInt16.self) else {
-            return .needMoreData
-        }
-        
-        if buffer.readableBytes >= len + 2 {
-            buffer.moveReaderIndex(forwardBy: 2)
-            if let bytes = buffer.readBytes(length: Int(len)) {
-                context.fireChannelRead(self.wrapInboundOut(ByteBuffer(bytes: bytes)))
-            }
-            return .continue
-        } else {
-            return .needMoreData
-        }
-    }
 }

Sources/Punchnet/SDLUDPHole.swift

@@ -8,101 +8,36 @@
 import Foundation
 import NIOCore
 import NIOPosix
+import Combine
 
 // 处理和sn-server服务器之间的通讯
-actor SDLUDPHole {
+class SDLUDPHole: ChannelInboundHandler, @unchecked Sendable {
+    public typealias InboundIn = AddressedEnvelope<ByteBuffer>
+    public typealias OutboundOut = AddressedEnvelope<ByteBuffer>
+    
+    // 回调函数
+    public typealias CallbackFun = (SDLStunProbeReply?) -> Void
+    
     private let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
-    private let asyncChannel: NIOAsyncChannel<AddressedEnvelope<ByteBuffer>, AddressedEnvelope<ByteBuffer>>
-    private let (writeStream, writeContinuation) = AsyncStream.makeStream(of: UDPMessage.self, bufferingPolicy: .unbounded)
     
     private var cookieGenerator = SDLIdGenerator(seed: 1)
-    private var promises: [UInt32:EventLoopPromise<SDLStunProbeReply>] = [:]
+    private var callbackManager = HoleCallbackManager()
+    
     public var localAddress: SocketAddress?
+    public var channel: Channel?
     
-    public let (eventFlow, eventContinuation) = AsyncStream.makeStream(of: UDPEvent.self, bufferingPolicy: .unbounded)
+    public var eventFlow = PassthroughSubject<UDPEvent, Never>()
-    
-    struct UDPMessage {
-        let remoteAddress: SocketAddress
-        let type: SDLPacketType
-        let data: Data
-    }
     
     // 定义事件类型
     enum UDPEvent {
         case ready
+        case closed
         case message(SocketAddress, SDLHoleInboundMessage)
         case data(SDLData)
     }
     
-    // 启动函数
+    init() {
-    init() async throws {
-        let bootstrap = DatagramBootstrap(group: group)
-            .channelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
         
-        self.asyncChannel = try await bootstrap.bind(host: "0.0.0.0", port: 0)
-            .flatMapThrowing { channel in
-                return try NIOAsyncChannel(wrappingChannelSynchronously: channel, configuration: .init(
-                    inboundType: AddressedEnvelope<ByteBuffer>.self,
-                    outboundType: AddressedEnvelope<ByteBuffer>.self
-                ))
-            }
-            .get()
-        
-        self.localAddress = self.asyncChannel.channel.localAddress
-        SDLLogger.log("[UDPHole] started and listening on: \(self.localAddress!)", level: .debug)
-    }
-    
-    func start() async throws {
-        try await self.asyncChannel.executeThenClose { inbound, outbound in
-            self.eventContinuation.yield(.ready)
-            
-            try await withThrowingTaskGroup(of: Void.self) { group in
-                group.addTask {
-                    try await self.asyncChannel.channel.closeFuture.get()
-                    NSLog("[UDPHole] channel closed")
-                    throw SDLError.socketClosed
-                }
-                
-                group.addTask {
-                    for try await envelope in inbound {
-                        var buffer = envelope.data
-                        let remoteAddress = envelope.remoteAddress
-                        do {
-                            if let message = try Self.decode(buffer: &buffer) {
-                                switch message {
-                                case .data(let data):
-                                    SDLLogger.log("[SDLUDPHole] read data: \(data.format()), from: \(remoteAddress)", level: .debug)
-                                    self.eventContinuation.yield(.data(data))
-                                case .stunProbeReply(let probeReply):
-                                    // 执行并移除回调
-                                    await self.trigger(probeReply: probeReply)
-                                default:
-                                    self.eventContinuation.yield(.message(remoteAddress, message))
-                                }
-                            } else {
-                                SDLLogger.log("[SDLUDPHole] decode message, get null", level: .warning)
-                            }
-                        } catch let err {
-                            SDLLogger.log("[SDLUDPHole] decode message, get error: \(err)", level: .debug)
-                            throw err
-                        }
-                    }
-                }
-                
-                group.addTask {
-                    for try await message in self.writeStream {
-                        var buffer = self.asyncChannel.channel.allocator.buffer(capacity: message.data.count + 1)
-                        buffer.writeBytes([message.type.rawValue])
-                        buffer.writeBytes(message.data)
-                        
-                        let envelope = AddressedEnvelope<ByteBuffer>(remoteAddress: message.remoteAddress, data: buffer)
-                        try await outbound.write(envelope)
-                    }
-                }
-                
-                try await group.waitForAll()
-            }
-        }
     }
     
     // MARK: super_node apis
@@ -127,33 +62,26 @@ actor SDLUDPHole {
     }
     
     // 探测tun信息
-    func stunProbe(remoteAddress: SocketAddress, attr: SDLProbeAttr = .none, timeout: Int = 5) async throws -> SDLStunProbeReply {
+    func stunProbe(remoteAddress: SocketAddress, attr: SDLProbeAttr = .none, timeout: Int = 5) async -> SDLStunProbeReply? {
-        return try await self._stunProbe(remoteAddress: remoteAddress, attr: attr, timeout: timeout).get()
+        return await withCheckedContinuation { continuation in
+            self.stunProbe(remoteAddress: remoteAddress, attr: attr, timeout: timeout) { probeReply in
+                continuation.resume(returning: probeReply)
+            }
+        }
     }
     
-    private func _stunProbe(remoteAddress: SocketAddress, attr: SDLProbeAttr = .none, timeout: Int) -> EventLoopFuture<SDLStunProbeReply> {
+    private func stunProbe(remoteAddress: SocketAddress, attr: SDLProbeAttr = .none, timeout: Int, callback: @escaping CallbackFun) {
         let cookie = self.cookieGenerator.nextId()
+        
         var stunProbe = SDLStunProbe()
         stunProbe.cookie = cookie
         stunProbe.attr = UInt32(attr.rawValue)
+        
         self.send(remoteAddress: remoteAddress, type: .stunProbe, data: try! stunProbe.serializedData())
+        
         SDLLogger.log("[SDLUDPHole] stunProbe: \(remoteAddress)", level: .warning)
         
-        let promise = self.asyncChannel.channel.eventLoop.makePromise(of: SDLStunProbeReply.self)
+        self.callbackManager.addCallback(id: cookie, callback: callback)
-        self.promises[cookie] = promise
-        
-        return promise.futureResult
-    }
-    
-    private func trigger(probeReply: SDLStunProbeReply) {
-        let id = probeReply.cookie
-        // 执行并移除回调
-        if let promise = self.promises[id] {
-            self.asyncChannel.channel.eventLoop.execute {
-                promise.succeed(probeReply)
-            }
-            self.promises.removeValue(forKey: id)
-        }
     }
     
     // MARK: client-client apis
@@ -168,10 +96,11 @@ actor SDLUDPHole {
         dataPacket.dstMac = session.dstMac
         dataPacket.ttl = 255
         dataPacket.data = data
-        if let packet = try? dataPacket.serializedData() {
+        let packet = try! dataPacket.serializedData()
-            SDLLogger.log("[SDLUDPHole] sendPacket: \(remoteAddress), count: \(packet.count)", level: .debug)
+        
-            self.send(remoteAddress: remoteAddress, type: .data, data: packet)
+        SDLLogger.log("[SDLUDPHole] sendPacket: \(remoteAddress), count: \(packet.count)", level: .debug)
-        }
+        
+        self.send(remoteAddress: remoteAddress, type: .data, data: packet)
     }
     
     // 通过sn服务器转发数据包, data已经是加密过后的数据
@@ -185,23 +114,23 @@ actor SDLUDPHole {
         dataPacket.ttl = 255
         dataPacket.data = data
         
-        if let packet = try? dataPacket.serializedData() {
+        let packet = try! dataPacket.serializedData()
-            NSLog("[SDLContext] forward packet, remoteAddress: \(remoteAddress), data size: \(packet.count)")
+        
-            self.send(remoteAddress: remoteAddress, type: .data, data: packet)
+        NSLog("[SDLContext] forward packet, remoteAddress: \(remoteAddress), data size: \(packet.count)")
-        }
+        
+        self.send(remoteAddress: remoteAddress, type: .data, data: packet)
     }
     
     // 发送register包
-    func sendRegister(remoteAddress: SocketAddress, networkId: UInt32, srcMac: Data, dst_mac: Data) {
+    func sendRegister(context ctx: SDLContext, remoteAddress: SocketAddress, dst_mac: Data) {
         var register = SDLRegister()
-        register.networkID = networkId
+        register.networkID = ctx.devAddr.networkID
-        register.srcMac = srcMac
+        register.srcMac = ctx.devAddr.mac
         register.dstMac = dst_mac
         
-        if let packet = try? register.serializedData() {
+        SDLLogger.log("[SDLUDPHole] SendRegister: \(remoteAddress), src_mac: \(LayerPacket.MacAddress.description(data: ctx.devAddr.mac)), dst_mac: \(LayerPacket.MacAddress.description(data: dst_mac))", level: .debug)
-            SDLLogger.log("[SDLUDPHole] SendRegister: \(remoteAddress), src_mac: \(LayerPacket.MacAddress.description(data: srcMac)), dst_mac: \(LayerPacket.MacAddress.description(data: dst_mac))", level: .debug)
+        
-            self.send(remoteAddress: remoteAddress, type: .register, data: packet)
+        self.send(remoteAddress: remoteAddress, type: .register, data: try! register.serializedData())
-        }
     }
     
     // 回复registerAck
@@ -211,20 +140,104 @@ actor SDLUDPHole {
         registerAck.srcMac = ctx.devAddr.mac
         registerAck.dstMac = dst_mac
         
-        if let packet = try? registerAck.serializedData() {
+        SDLLogger.log("[SDLUDPHole] SendRegisterAck: \(remoteAddress), \(registerAck)", level: .debug)
-            SDLLogger.log("[SDLUDPHole] SendRegisterAck: \(remoteAddress), \(registerAck)", level: .debug)
+        
-            self.send(remoteAddress: remoteAddress, type: .registerAck, data: packet)
+        self.send(remoteAddress: remoteAddress, type: .registerAck, data: try! registerAck.serializedData())
+    }
+    
+    // 启动函数
+    func start() async throws {
+        let bootstrap = DatagramBootstrap(group: self.group)
+            .channelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
+            .channelInitializer { channel in
+                // 接收缓冲区
+                return channel.setOption(ChannelOptions.socketOption(.so_rcvbuf), value: 5 * 1024  * 1024)
+                    .flatMap {
+                        channel.setOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_SNDBUF), value: 5 * 1024  * 1024)
+                    }.flatMap {
+                        channel.pipeline.addHandler(self)
+                    }
+            }
+        
+        let channel = try await bootstrap.bind(host: "0.0.0.0", port: 0).get()
+        
+        SDLLogger.log("[UDPHole] started and listening on: \(channel.localAddress!)", level: .debug)
+        self.localAddress = channel.localAddress
+        self.channel = channel
+    }
+    
+    // -- MARK: ChannelInboundHandler Methods
+    
+    public func channelActive(context: ChannelHandlerContext) {
+        self.eventFlow.send(.ready)
+    }
+    
+    // 接收到的消息, 消息需要根据类型分流
+    public func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+        let envelope = self.unwrapInboundIn(data)
+        var buffer = envelope.data
+        let remoteAddress = envelope.remoteAddress
+        
+        do {
+            if let message = try decode(buffer: &buffer) {
+                Task {
+                    switch message {
+                    case .data(let data):
+                        SDLLogger.log("[SDLUDPHole] read data: \(data.format()), from: \(remoteAddress)", level: .debug)
+                        self.eventFlow.send(.data(data))
+                    case .stunProbeReply(let probeReply):
+                        self.callbackManager.fireCallback(message: probeReply)
+                    default:
+                        self.eventFlow.send(.message(remoteAddress, message))
+                    }
+                }
+            } else {
+                SDLLogger.log("[SDLUDPHole] decode message, get null", level: .warning)
+            }
+        } catch let err {
+            SDLLogger.log("[SDLUDPHole] decode message, get error: \(err)", level: .debug)
         }
     }
     
-    // 处理写入逻辑
+    public func errorCaught(context: ChannelHandlerContext, error: Error) {
-    private func send(remoteAddress: SocketAddress, type: SDLPacketType, data: Data) {
+        SDLLogger.log("[SDLUDPHole] get error: \(error)", level: .error)
-        let message = UDPMessage(remoteAddress: remoteAddress, type: type, data: data)
+        // As we are not really interested getting notified on success or failure we just pass nil as promise to
-        self.writeContinuation.yield(message)
+        // reduce allocations.
+        context.close(promise: nil)
+        self.channel = nil
+        self.eventFlow.send(.closed)
     }
     
-    //--MARK: 编解码器
+    public func channelInactive(context: ChannelHandlerContext) {
-    private static func decode(buffer: inout ByteBuffer) throws -> SDLHoleInboundMessage? {
+        self.channel = nil
+        context.close(promise: nil)
+    }
+    
+    // 处理写入逻辑
+    func send(remoteAddress: SocketAddress, type: SDLPacketType, data: Data) {
+        guard let channel = self.channel else {
+            return
+        }
+        
+        channel.eventLoop.execute {
+            var buffer = channel.allocator.buffer(capacity: data.count + 1)
+            buffer.writeBytes([type.rawValue])
+            buffer.writeBytes(data)
+            
+            let envelope = AddressedEnvelope<ByteBuffer>(remoteAddress: remoteAddress, data: buffer)
+            channel.writeAndFlush(self.wrapOutboundOut(envelope), promise: nil)
+        }
+    }
+    
+    deinit {
+        try? self.group.syncShutdownGracefully()
+    }
+}
+
+//--MARK: 编解码器
+extension SDLUDPHole {
+    
+    func decode(buffer: inout ByteBuffer) throws -> SDLHoleInboundMessage? {
         guard let type = buffer.readInteger(as: UInt8.self),
               let packetType = SDLPacketType(rawValue: type),
               let bytes = buffer.readBytes(length: buffer.readableBytes) else {
@@ -252,10 +265,42 @@ actor SDLUDPHole {
             return nil
         }
     }
-    
+}
-    deinit {
+
-        try? self.group.syncShutdownGracefully()
+// --MARK: 回调函数管理器
-        self.writeContinuation.finish()
+extension SDLUDPHole {
-        self.eventContinuation.finish()
+    
-    }
+    private struct HoleCallbackManager {
+        // 存储回调函数和对应的超时任务
+        private var callbacks: [UInt32: CallbackFun] = [:]
+        
+        //private var timeoutCallbacks: [UInt32: CallbackFun] = [:]
+
+        // 添加回调并设置超时
+        mutating func addCallback(id: UInt32, callback: @escaping CallbackFun) {
+            // 存储回调
+            self.callbacks[id] = callback
+        }
+        
+        // 正常触发回调（收到响应）
+        mutating func fireCallback(message: SDLStunProbeReply) {
+            let id = message.cookie
+            // 执行并移除回调
+            if let callback = callbacks[id] {
+                callback(message)
+                self.callbacks.removeValue(forKey: id)
+            }
+        }
+        
+        // 触发所有回调（清理场景）
+        mutating func fireAllCallbacks(message: SDLSuperInboundMessage) {
+            // 触发所有回调
+            for callback in callbacks.values {
+                callback(nil)
+            }
+            self.callbacks.removeAll()
+        }
+        
+    }
+    
 }