fix actor
This commit is contained in:
parent
1f629a58a0
commit
9a76b61ff0
@ -47,8 +47,8 @@ public class SDLContext: @unchecked Sendable {
|
||||
let rsaCipher: RSACipher
|
||||
|
||||
// 依赖的变量
|
||||
var udpHoleActor: SDLUDPHole?
|
||||
var superClientActor: SDLSuperClient?
|
||||
var udpHole: SDLUDPHole?
|
||||
var superClient: SDLSuperClient?
|
||||
|
||||
// 数据包读取任务
|
||||
private var readTask: Task<(), Never>?
|
||||
@ -93,17 +93,17 @@ public class SDLContext: @unchecked Sendable {
|
||||
}
|
||||
|
||||
public func start() async throws {
|
||||
self.udpHoleActor = try await SDLUDPHole()
|
||||
self.superClientActor = try await SDLSuperClient(host: self.config.superHost, port: self.config.superPort)
|
||||
self.udpHole = try await SDLUDPHole()
|
||||
self.superClient = try await SDLSuperClient(host: self.config.superHost, port: self.config.superPort)
|
||||
self.noticeClient = try await SDLNoticeClient()
|
||||
|
||||
try await withThrowingTaskGroup(of: Void.self) { group in
|
||||
group.addTask {
|
||||
try await self.udpHoleActor?.start()
|
||||
try await self.udpHole?.start()
|
||||
}
|
||||
|
||||
group.addTask {
|
||||
try await self.superClientActor?.start()
|
||||
try await self.superClient?.start()
|
||||
}
|
||||
|
||||
group.addTask {
|
||||
@ -111,7 +111,7 @@ public class SDLContext: @unchecked Sendable {
|
||||
}
|
||||
|
||||
group.addTask {
|
||||
if let eventFlow = self.superClientActor?.eventFlow {
|
||||
if let eventFlow = self.superClient?.eventFlow {
|
||||
for try await event in eventFlow {
|
||||
try await self.handleSuperEvent(event: event)
|
||||
}
|
||||
@ -119,7 +119,7 @@ public class SDLContext: @unchecked Sendable {
|
||||
}
|
||||
|
||||
group.addTask {
|
||||
if let eventFlow = self.udpHoleActor?.eventFlow {
|
||||
if let eventFlow = self.udpHole?.eventFlow {
|
||||
for try await event in eventFlow {
|
||||
try await self.handleUDPEvent(event: event)
|
||||
}
|
||||
@ -130,32 +130,32 @@ public class SDLContext: @unchecked Sendable {
|
||||
group.addTask {
|
||||
while !Task.isCancelled {
|
||||
try await Task.sleep(nanoseconds: 5 * 1_000_000_000)
|
||||
self.lastCookie = await self.udpHoleActor?.stunRequest(context: self)
|
||||
self.lastCookie = await self.udpHole?.stunRequest(context: self)
|
||||
}
|
||||
}
|
||||
|
||||
try await group.waitForAll()
|
||||
}
|
||||
|
||||
// // 启动网络监控
|
||||
// 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()
|
||||
// 启动网络监控
|
||||
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()
|
||||
}
|
||||
|
||||
public func stop() async {
|
||||
self.superClientActor = nil
|
||||
self.udpHoleActor = nil
|
||||
self.superClient = nil
|
||||
self.udpHole = nil
|
||||
|
||||
self.readTask?.cancel()
|
||||
}
|
||||
@ -164,7 +164,7 @@ public class SDLContext: @unchecked Sendable {
|
||||
switch event {
|
||||
case .ready:
|
||||
NSLog("[SDLContext] get registerSuper, mac address: \(Self.formatMacAddress(mac: self.devAddr.mac))")
|
||||
guard let message = try await self.superClientActor?.registerSuper(context: self) else {
|
||||
guard let message = try await self.superClient?.registerSuper(context: self) else {
|
||||
return
|
||||
}
|
||||
|
||||
@ -231,7 +231,7 @@ 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.udpHoleActor?.sendRegister(context: self, remoteAddress: remoteAddress, dst_mac: sendRegisterEvent.dstMac)
|
||||
await self.udpHole?.sendRegister(context: self, remoteAddress: remoteAddress, dst_mac: sendRegisterEvent.dstMac)
|
||||
}
|
||||
|
||||
case .networkShutdown(let shutdownEvent):
|
||||
@ -254,7 +254,7 @@ public class SDLContext: @unchecked Sendable {
|
||||
var commandAck = SDLCommandAck()
|
||||
commandAck.status = true
|
||||
|
||||
await self.superClientActor?.commandAck(packetId: packetId, ack: commandAck)
|
||||
await self.superClient?.commandAck(packetId: packetId, ack: commandAck)
|
||||
}
|
||||
}
|
||||
|
||||
@ -294,7 +294,7 @@ public class SDLContext: @unchecked Sendable {
|
||||
// 判断目标地址是否是tun的网卡地址, 并且是在同一个网络下
|
||||
if register.dstMac == self.devAddr.mac && register.networkID == self.devAddr.networkID {
|
||||
// 回复ack包
|
||||
await self.udpHoleActor?.sendRegisterAck(context: self, remoteAddress: remoteAddress, dst_mac: register.srcMac)
|
||||
await 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)
|
||||
@ -383,7 +383,7 @@ public class SDLContext: @unchecked Sendable {
|
||||
.sink { _ in
|
||||
Task {
|
||||
let (forwardNum, p2pNum, inboundNum) = await self.flowTracer.reset()
|
||||
await self.superClientActor?.flowReport(forwardNum: forwardNum, p2pNum: p2pNum, inboundNum: inboundNum)
|
||||
await self.superClient?.flowReport(forwardNum: forwardNum, p2pNum: p2pNum, inboundNum: inboundNum)
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -483,13 +483,13 @@ public class SDLContext: @unchecked Sendable {
|
||||
// 通过session发送到对端
|
||||
if let session = await self.sessionManager.getSession(toAddress: dstMac) {
|
||||
NSLog("[SDLContext] send packet by session: \(session)")
|
||||
await self.udpHoleActor?.sendPacket(context: self, session: session, data: encodedPacket)
|
||||
await self.udpHole?.sendPacket(context: self, session: session, data: encodedPacket)
|
||||
|
||||
await self.flowTracer.inc(num: data.count, type: .p2p)
|
||||
}
|
||||
else {
|
||||
// 通过super_node进行转发
|
||||
await self.udpHoleActor?.forwardPacket(context: self, dst_mac: dstMac, data: encodedPacket)
|
||||
await self.udpHole?.forwardPacket(context: self, dst_mac: dstMac, data: encodedPacket)
|
||||
// 流量统计
|
||||
await self.flowTracer.inc(num: data.count, type: .forward)
|
||||
|
||||
@ -502,7 +502,7 @@ public class SDLContext: @unchecked Sendable {
|
||||
|
||||
func holerTask(dstMac: Data) -> Task<(), Never> {
|
||||
return Task {
|
||||
guard let message = try? await self.superClientActor?.queryInfo(context: self, dst_mac: dstMac) else {
|
||||
guard let message = try? await self.superClient?.queryInfo(context: self, dst_mac: dstMac) else {
|
||||
return
|
||||
}
|
||||
|
||||
@ -513,7 +513,7 @@ public class SDLContext: @unchecked Sendable {
|
||||
if let remoteAddress = peerInfo.v4Info.socketAddress() {
|
||||
SDLLogger.log("[SDLContext] hole sock address: \(remoteAddress)", level: .warning)
|
||||
// 发送register包
|
||||
await self.udpHoleActor?.sendRegister(context: self, remoteAddress: remoteAddress, dst_mac: dstMac)
|
||||
await self.udpHole?.sendRegister(context: self, remoteAddress: remoteAddress, dst_mac: dstMac)
|
||||
} else {
|
||||
SDLLogger.log("[SDLContext] hole sock address is invalid: \(peerInfo.v4Info)", level: .warning)
|
||||
}
|
||||
@ -525,8 +525,8 @@ public class SDLContext: @unchecked Sendable {
|
||||
|
||||
deinit {
|
||||
self.stunCancel?.cancel()
|
||||
self.udpHoleActor = nil
|
||||
self.superClientActor = nil
|
||||
self.udpHole = nil
|
||||
self.superClient = nil
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user