fix tick worker
This commit is contained in:
parent
1c1bec9d78
commit
153f172daf
@ -132,13 +132,14 @@ public actor PeriodicWorker {
|
||||
}
|
||||
|
||||
private static func runFixedRate(configuration: Configuration, operation: @escaping Operation, onError: @escaping ErrorHandler) async {
|
||||
var nextRun = ContinuousClock.now
|
||||
let clock = ContinuousClock()
|
||||
var nextRun = clock.now
|
||||
if !configuration.runImmediately {
|
||||
nextRun = nextRun.advanced(by: configuration.interval)
|
||||
}
|
||||
|
||||
while !Task.isCancelled {
|
||||
let now = ContinuousClock.now
|
||||
let now = clock.now
|
||||
if nextRun > now {
|
||||
let delay = now.duration(to: nextRun)
|
||||
guard await sleep(interval: delay, tolerance: configuration.tolerance) else {
|
||||
@ -155,7 +156,7 @@ public actor PeriodicWorker {
|
||||
return
|
||||
}
|
||||
|
||||
let afterRun = ContinuousClock.now
|
||||
let afterRun = clock.now
|
||||
repeat {
|
||||
nextRun = nextRun.advanced(by: configuration.interval)
|
||||
} while nextRun <= afterRun
|
||||
|
||||
@ -90,11 +90,11 @@ actor SDLContextActor {
|
||||
nonisolated private let provider: NEPacketTunnelProvider
|
||||
|
||||
// 处理权限控制
|
||||
private var updatePolicyTask: Task<Void, Never>?
|
||||
private let policyService: PolicyService
|
||||
private var updatePolicyWorker: PeriodicWorker?
|
||||
|
||||
// stunRequest任务
|
||||
private var stunRequestTask: Task<Void, Never>?
|
||||
private var stunRequestWorker: PeriodicWorker?
|
||||
|
||||
public init(provider: NEPacketTunnelProvider, config: SDLConfiguration, rsaCipher: RSACipher) {
|
||||
let puncherActor = SDLPuncherActor()
|
||||
@ -183,11 +183,11 @@ actor SDLContextActor {
|
||||
await self.arpServer.stop()
|
||||
await self.sessionManager.clear()
|
||||
|
||||
self.stunRequestTask?.cancel()
|
||||
self.stunRequestTask = nil
|
||||
await self.stunRequestWorker?.stop()
|
||||
self.stunRequestWorker = nil
|
||||
|
||||
self.updatePolicyTask?.cancel()
|
||||
self.updatePolicyTask = nil
|
||||
await self.updatePolicyWorker?.stop()
|
||||
self.updatePolicyWorker = nil
|
||||
await self.policyService.clear()
|
||||
|
||||
await self.packetOutboundActor.stop()
|
||||
@ -349,19 +349,27 @@ extension SDLContextActor {
|
||||
|
||||
// 注册成功super的回调函数
|
||||
private func whenRegistedSuper() async {
|
||||
self.updatePolicyTask?.cancel()
|
||||
await self.updatePolicyWorker?.stop()
|
||||
let policyService = self.policyService
|
||||
let superServiceProxy = self.superServiceProxy
|
||||
|
||||
self.updatePolicyTask = Task {
|
||||
do {
|
||||
while true {
|
||||
try await Task.sleep(for: .seconds(300))
|
||||
let updatePolicyWorker = PeriodicWorker(
|
||||
configuration: .init(
|
||||
interval: .seconds(300),
|
||||
runImmediately: false,
|
||||
mode: .fixedDelay,
|
||||
errorPolicy: .keepRunning(delay: .seconds(5))
|
||||
),
|
||||
operation: {
|
||||
SDLLogger.log("[SDLContext] updatePolicyTask execute")
|
||||
await self.policyService.updatePolicy(superServiceProxy: self.superServiceProxy)
|
||||
}
|
||||
} catch let err {
|
||||
await policyService.updatePolicy(superServiceProxy: superServiceProxy)
|
||||
},
|
||||
onError: { err in
|
||||
SDLLogger.log("[SDLContext] updatePolicyTask stop with err: \(err)")
|
||||
}
|
||||
}
|
||||
)
|
||||
self.updatePolicyWorker = updatePolicyWorker
|
||||
await updatePolicyWorker.start()
|
||||
|
||||
// 启动stun任务
|
||||
await self.startStunRequestTask()
|
||||
@ -519,23 +527,17 @@ extension SDLContextActor {
|
||||
|
||||
// MARK: -- StunRequestTask
|
||||
private func startStunRequestTask() async {
|
||||
self.stunRequestTask?.cancel()
|
||||
self.stunRequestTask = nil
|
||||
|
||||
// 处理心跳逻辑
|
||||
self.stunRequestTask = Task { [weak self] in
|
||||
let timerStream = SDLAsyncTimerStream()
|
||||
timerStream.start(interval: .seconds(8))
|
||||
|
||||
for await _ in timerStream.stream {
|
||||
if Task.isCancelled {
|
||||
break
|
||||
}
|
||||
await self.stunRequestWorker?.stop()
|
||||
|
||||
let stunRequestWorker = PeriodicWorker(
|
||||
configuration: .init(
|
||||
interval: .seconds(8),
|
||||
runImmediately: true,
|
||||
mode: .fixedDelay,
|
||||
errorPolicy: .keepRunning(delay: .seconds(5))
|
||||
),
|
||||
operation: { [weak self] in
|
||||
let probeReply = try? await self?.ipv6AssistClient?.probe(requestTimeout: .seconds(3))
|
||||
if Task.isCancelled {
|
||||
break
|
||||
}
|
||||
|
||||
if let v6Info = probeReply?.v6Info, let v6Address = SDLUtil.ipv6DataToString(v6Info.v6) {
|
||||
SDLLogger.log("[SDLContext] probe ipv6 address: \(v6Address)")
|
||||
@ -543,15 +545,14 @@ extension SDLContextActor {
|
||||
SDLLogger.log("[SDLContext] probe ipv6 address: empty")
|
||||
}
|
||||
|
||||
if Task.isCancelled {
|
||||
break
|
||||
}
|
||||
|
||||
await self?.sendStunRequest(v6Info: probeReply?.v6Info)
|
||||
},
|
||||
onError: { err in
|
||||
SDLLogger.log("[SDLContext] udp stunRequestTask stop with err: \(err)")
|
||||
}
|
||||
|
||||
SDLLogger.log("[SDLContext] udp stunRequestTask cancel")
|
||||
}
|
||||
)
|
||||
self.stunRequestWorker = stunRequestWorker
|
||||
await stunRequestWorker.start()
|
||||
}
|
||||
|
||||
private func sendStunRequest(v6Info: SDLV6Info?) async {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user