fix
This commit is contained in:
parent
2f9920ad6d
commit
478969d99d
@ -23,7 +23,7 @@ actor SDLContextActor {
|
||||
nonisolated let aesCipher: AESCipher
|
||||
|
||||
// aes
|
||||
var aesKey: Data = Data()
|
||||
private var aesKey: Data?
|
||||
|
||||
// rsa的相关配置, public_key是本地生成的
|
||||
nonisolated let rsaCipher: RSACipher
|
||||
@ -127,6 +127,9 @@ actor SDLContextActor {
|
||||
self.logger.log("[SDLContext] udpHole started")
|
||||
self.udpHole = udpHole
|
||||
|
||||
await udpHole.channelIsActived()
|
||||
await self.handleUDPHoleReady()
|
||||
|
||||
try await withThrowingTaskGroup(of: Void.self) { group in
|
||||
group.addTask {
|
||||
try await channel.closeFuture.get()
|
||||
@ -142,19 +145,6 @@ actor SDLContextActor {
|
||||
}
|
||||
}
|
||||
|
||||
// 处理event事件流
|
||||
group.addTask {
|
||||
for try await event in udpHole.eventStream {
|
||||
try Task.checkCancellation()
|
||||
switch event {
|
||||
case .ready:
|
||||
await self.handleUDPHoleReady()
|
||||
case .closed:
|
||||
()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 处理数据流
|
||||
group.addTask {
|
||||
for try await data in udpHole.dataStream {
|
||||
@ -203,39 +193,39 @@ actor SDLContextActor {
|
||||
self.natType = natType
|
||||
}
|
||||
|
||||
// 处理和super的协商问题
|
||||
private func handleUDPHoleReady() async {
|
||||
if let udpHole = self.udpHole {
|
||||
self.puncherActor = SDLPuncherActor(udpHole: udpHole, querySocketAddress: config.stunSocketAddress, logger: logger)
|
||||
self.proberActor = SDLNATProberActor(udpHole: udpHole, addressArray: self.config.stunProbeSocketAddressArray, logger: self.logger)
|
||||
guard let udpHole = self.udpHole else {
|
||||
return
|
||||
}
|
||||
|
||||
await withDiscardingTaskGroup { group in
|
||||
group.addTask {
|
||||
// 开始探测nat的类型
|
||||
if let natType = await self.proberActor?.probeNatType() {
|
||||
await self.setNatType(natType: natType)
|
||||
self.logger.log("[SDLContext] nat_type is: \(natType)")
|
||||
}
|
||||
}
|
||||
self.puncherActor = SDLPuncherActor(udpHole: udpHole, querySocketAddress: config.stunSocketAddress, logger: logger)
|
||||
self.proberActor = SDLNATProberActor(udpHole: udpHole, addressArray: self.config.stunProbeSocketAddressArray, logger: self.logger)
|
||||
|
||||
group.addTask {
|
||||
var registerSuper = SDLRegisterSuper()
|
||||
registerSuper.pktID = 0
|
||||
registerSuper.clientID = self.config.clientId
|
||||
registerSuper.networkID = self.config.networkAddress.networkId
|
||||
registerSuper.mac = self.config.networkAddress.mac
|
||||
registerSuper.ip = self.config.networkAddress.ip
|
||||
registerSuper.maskLen = UInt32(self.config.networkAddress.maskLen)
|
||||
registerSuper.hostname = self.config.hostname
|
||||
registerSuper.pubKey = self.rsaCipher.pubKey
|
||||
registerSuper.accessToken = self.config.accessToken
|
||||
|
||||
if let registerSuperData = try? registerSuper.serializedData() {
|
||||
self.logger.log("[SDLContext] will send register super")
|
||||
await self.udpHole?.send(type: .registerSuper, data: registerSuperData, remoteAddress: self.config.stunSocketAddress)
|
||||
}
|
||||
// 开始探测nat的类型
|
||||
Task.detached {
|
||||
if let natType = await self.proberActor?.probeNatType() {
|
||||
await self.setNatType(natType: natType)
|
||||
self.logger.log("[SDLContext] nat_type is: \(natType)")
|
||||
}
|
||||
}
|
||||
|
||||
// 注册
|
||||
var registerSuper = SDLRegisterSuper()
|
||||
registerSuper.pktID = 0
|
||||
registerSuper.clientID = self.config.clientId
|
||||
registerSuper.networkID = self.config.networkAddress.networkId
|
||||
registerSuper.mac = self.config.networkAddress.mac
|
||||
registerSuper.ip = self.config.networkAddress.ip
|
||||
registerSuper.maskLen = UInt32(self.config.networkAddress.maskLen)
|
||||
registerSuper.hostname = self.config.hostname
|
||||
registerSuper.pubKey = self.rsaCipher.pubKey
|
||||
registerSuper.accessToken = self.config.accessToken
|
||||
|
||||
if let registerSuperData = try? registerSuper.serializedData() {
|
||||
self.logger.log("[SDLContext] will send register super")
|
||||
self.udpHole?.send(type: .registerSuper, data: registerSuperData, remoteAddress: self.config.stunSocketAddress)
|
||||
}
|
||||
}
|
||||
|
||||
private func sendStunRequest() async {
|
||||
@ -348,6 +338,10 @@ actor SDLContextActor {
|
||||
}
|
||||
|
||||
private func handleData(data: SDLData) throws {
|
||||
guard let aesKey = self.aesKey else {
|
||||
return
|
||||
}
|
||||
|
||||
let mac = LayerPacket.MacAddress(data: data.dstMac)
|
||||
|
||||
let networkAddr = config.networkAddress
|
||||
@ -355,7 +349,7 @@ actor SDLContextActor {
|
||||
return
|
||||
}
|
||||
|
||||
guard let decyptedData = try? self.aesCipher.decypt(aesKey: self.aesKey, data: Data(data.data)) else {
|
||||
guard let decyptedData = try? self.aesCipher.decypt(aesKey: aesKey, data: Data(data.data)) else {
|
||||
return
|
||||
}
|
||||
|
||||
@ -477,7 +471,7 @@ actor SDLContextActor {
|
||||
let networkAddr = self.config.networkAddress
|
||||
// 将数据封装层2层的数据包
|
||||
let layerPacket = LayerPacket(dstMac: dstMac, srcMac: networkAddr.mac, type: type, data: data)
|
||||
guard let encodedPacket = try? self.aesCipher.encrypt(aesKey: self.aesKey, data: layerPacket.marshal()) else {
|
||||
guard let aesKey = self.aesKey, let encodedPacket = try? self.aesCipher.encrypt(aesKey: aesKey, data: layerPacket.marshal()) else {
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@ -31,8 +31,9 @@ final class SDLUDPHole: ChannelInboundHandler {
|
||||
public let dataStream: AsyncStream<SDLData>
|
||||
private let dataContinuation: AsyncStream<SDLData>.Continuation
|
||||
|
||||
public let eventStream: AsyncStream<HoleEvent>
|
||||
private let eventContinuation: AsyncStream<HoleEvent>.Continuation
|
||||
// 解决channelready的问题
|
||||
private var cont: CheckedContinuation<Void, Never>?
|
||||
private var isReady: Bool = false
|
||||
|
||||
private let logger: SDLLogger
|
||||
|
||||
@ -46,10 +47,9 @@ final class SDLUDPHole: ChannelInboundHandler {
|
||||
self.logger = logger
|
||||
(self.signalStream, self.signalContinuation) = AsyncStream.makeStream(of: (SocketAddress, SDLHoleSignal).self, bufferingPolicy: .unbounded)
|
||||
(self.dataStream, self.dataContinuation) = AsyncStream.makeStream(of: SDLData.self, bufferingPolicy: .unbounded)
|
||||
(self.eventStream, self.eventContinuation) = AsyncStream.makeStream(of: HoleEvent.self, bufferingPolicy: .unbounded)
|
||||
}
|
||||
|
||||
func start() throws -> Channel {
|
||||
func start() throws -> Channel {
|
||||
let bootstrap = DatagramBootstrap(group: group)
|
||||
.channelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
|
||||
.channelInitializer { channel in
|
||||
@ -63,10 +63,25 @@ final class SDLUDPHole: ChannelInboundHandler {
|
||||
return channel
|
||||
}
|
||||
|
||||
func channelIsActived() async {
|
||||
await withCheckedContinuation { c in
|
||||
if isReady {
|
||||
c.resume()
|
||||
} else {
|
||||
self.cont = c
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// --MARK: ChannelInboundHandler delegate
|
||||
|
||||
func channelActive(context: ChannelHandlerContext) {
|
||||
self.eventContinuation.yield(.ready)
|
||||
guard !isReady else {
|
||||
return
|
||||
}
|
||||
self.isReady = true
|
||||
self.cont?.resume()
|
||||
self.cont = nil
|
||||
}
|
||||
|
||||
func channelRead(context: ChannelHandlerContext, data: NIOAny) {
|
||||
@ -93,8 +108,6 @@ final class SDLUDPHole: ChannelInboundHandler {
|
||||
func channelInactive(context: ChannelHandlerContext) {
|
||||
self.signalContinuation.finish()
|
||||
self.dataContinuation.finish()
|
||||
self.eventContinuation.yield(.closed)
|
||||
self.eventContinuation.finish()
|
||||
|
||||
context.close(promise: nil)
|
||||
}
|
||||
@ -196,7 +209,6 @@ final class SDLUDPHole: ChannelInboundHandler {
|
||||
|
||||
deinit {
|
||||
try? self.group.syncShutdownGracefully()
|
||||
self.eventContinuation.finish()
|
||||
self.channel?.close(promise: nil)
|
||||
}
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user