134 lines
4.2 KiB
Swift
134 lines
4.2 KiB
Swift
//
|
||
// SDLUDPHoleV6.swift
|
||
// Tun
|
||
//
|
||
// Created by 安礼成 on 2026/4/15.
|
||
//
|
||
|
||
import Foundation
|
||
import NIOCore
|
||
import NIOPosix
|
||
import SwiftProtobuf
|
||
|
||
// 处理和sn-server服务器之间的通讯
|
||
final class SDLUDPHoleV6: ChannelInboundHandler {
|
||
typealias InboundIn = AddressedEnvelope<ByteBuffer>
|
||
|
||
// 事件
|
||
enum HoleEvent {
|
||
case ready
|
||
case closed
|
||
case errorCaught
|
||
}
|
||
|
||
private var isStopped: Bool = false
|
||
|
||
private let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
|
||
private var channel: Channel?
|
||
|
||
public let messageStream: AsyncStream<(SocketAddress, SDLHoleMessage)>
|
||
private let messageContinuation: AsyncStream<(SocketAddress, SDLHoleMessage)>.Continuation
|
||
|
||
// 事件相关逻辑
|
||
public let eventStream: AsyncStream<HoleEvent>
|
||
private let eventContinuation: AsyncStream<HoleEvent>.Continuation
|
||
|
||
// 启动函数
|
||
init() throws {
|
||
let (stream, continuation) = AsyncStream.makeStream(of: (SocketAddress, SDLHoleMessage).self, bufferingPolicy: .bufferingNewest(2048))
|
||
self.messageStream = stream
|
||
self.messageContinuation = continuation
|
||
|
||
let eventPair = AsyncStream.makeStream(of: HoleEvent.self)
|
||
self.eventStream = eventPair.stream
|
||
self.eventContinuation = eventPair.continuation
|
||
}
|
||
|
||
func start() throws -> SocketAddress? {
|
||
let bootstrap = DatagramBootstrap(group: group)
|
||
.channelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
|
||
.channelInitializer { channel in
|
||
channel.pipeline.addHandler(self)
|
||
}
|
||
|
||
// 绑定到IPv6通配地址，只处理IPv6流量
|
||
let channel = try bootstrap.bind(host: "::", port: 0).wait()
|
||
self.channel = channel
|
||
|
||
return channel.localAddress
|
||
}
|
||
|
||
// --MARK: ChannelInboundHandler delegate
|
||
|
||
func channelRead(context: ChannelHandlerContext, data: NIOAny) {
|
||
let envelope = unwrapInboundIn(data)
|
||
|
||
var buffer = envelope.data
|
||
let remoteAddress = envelope.remoteAddress
|
||
|
||
if let rawBytes = buffer.getBytes(at: buffer.readerIndex, length: buffer.readableBytes) {
|
||
SDLLogger.log("[SDLUDPHoleV6] get raw bytes: \(rawBytes.count), from: \(remoteAddress)", for: .debug)
|
||
}
|
||
|
||
do {
|
||
if let message = try SDLHoleMessage.decode(buffer: &buffer) {
|
||
self.messageContinuation.yield((remoteAddress, message))
|
||
} else {
|
||
SDLLogger.log("[SDLUDPHoleV6] decode message, get null", for: .debug)
|
||
}
|
||
} catch let err {
|
||
SDLLogger.log("[SDLUDPHoleV6] decode message, get error: \(err)", for: .debug)
|
||
}
|
||
}
|
||
|
||
func channelInactive(context: ChannelHandlerContext) {
|
||
SDLLogger.log("[SDLUDPHoleV6] channelInactive", for: .debug)
|
||
self.eventContinuation.yield(.closed)
|
||
}
|
||
|
||
func errorCaught(context: ChannelHandlerContext, error: any Error) {
|
||
SDLLogger.log("[SDLUDPHoleV6] channel error: \(error)", for: .debug)
|
||
context.close(promise: nil)
|
||
self.eventContinuation.yield(.errorCaught)
|
||
}
|
||
|
||
// MARK: 处理写入逻辑
|
||
func send(type: SDLPacketType, data: Data, remoteAddress: SocketAddress) {
|
||
guard let channel = self.channel else {
|
||
return
|
||
}
|
||
|
||
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.eventLoop.submit {
|
||
channel.writeAndFlush(envelope, promise: nil)
|
||
}
|
||
}
|
||
|
||
func stop() {
|
||
guard !self.isStopped else {
|
||
return
|
||
}
|
||
|
||
self.isStopped = true
|
||
SDLLogger.log("[SDLUDPHoleV6] stop", for: .debug)
|
||
|
||
self.messageContinuation.finish()
|
||
self.eventContinuation.finish()
|
||
|
||
let channel = self.channel
|
||
self.channel = nil
|
||
try? channel?.close().wait()
|
||
|
||
try? self.group.syncShutdownGracefully()
|
||
}
|
||
|
||
deinit {
|
||
SDLLogger.log("[SDLUDPHoleV6] deinit", for: .debug)
|
||
}
|
||
|
||
}
|