diff --git a/Tun/Punchnet/AsyncOneShot.swift b/Tun/Punchnet/Concurrency/AsyncOneShot.swift similarity index 100% rename from Tun/Punchnet/AsyncOneShot.swift rename to Tun/Punchnet/Concurrency/AsyncOneShot.swift diff --git a/Tun/Punchnet/OnceContinuation.swift b/Tun/Punchnet/Concurrency/OnceContinuation.swift similarity index 100% rename from Tun/Punchnet/OnceContinuation.swift rename to Tun/Punchnet/Concurrency/OnceContinuation.swift diff --git a/Tun/Punchnet/Concurrency/PeriodicWorker.swift b/Tun/Punchnet/Concurrency/PeriodicWorker.swift new file mode 100644 index 0000000..2398a21 --- /dev/null +++ b/Tun/Punchnet/Concurrency/PeriodicWorker.swift @@ -0,0 +1,197 @@ +// +// PeriodicWorker.swift +// Tun +// +// Created by Codex on 2026/5/20. +// + +import Foundation + +public actor PeriodicWorker { + public typealias Operation = @Sendable () async throws -> Void + public typealias ErrorHandler = @Sendable (Error) async -> Void + + public enum Mode: Sendable { + case fixedDelay + case fixedRate + } + + public enum ErrorPolicy: Sendable { + case keepRunning(delay: Duration?) + case stop + } + + public struct Configuration: Sendable { + public var interval: Duration + public var tolerance: Duration? + public var runImmediately: Bool + public var mode: Mode + public var errorPolicy: ErrorPolicy + + public init( + interval: Duration, + tolerance: Duration? = nil, + runImmediately: Bool = true, + mode: Mode = .fixedDelay, + errorPolicy: ErrorPolicy = .keepRunning(delay: .seconds(5)) + ) { + self.interval = interval + self.tolerance = tolerance + self.runImmediately = runImmediately + self.mode = mode + self.errorPolicy = errorPolicy + } + } + + private let configuration: Configuration + private let operation: Operation + private let onError: ErrorHandler + + private var task: Task? + private var generation: UInt64 = 0 + + public init(configuration: Configuration, operation: @escaping Operation, onError: @escaping ErrorHandler = { _ in }) { + self.configuration = configuration + self.operation = operation + self.onError = onError + } + + public func start() { + guard self.task == nil else { + return + } + + self.generation &+= 1 + let currentGeneration = self.generation + let configuration = self.configuration + let operation = self.operation + let onError = self.onError + + self.task = Task { + switch configuration.mode { + case .fixedDelay: + await Self.runFixedDelay( + configuration: configuration, + operation: operation, + onError: onError + ) + case .fixedRate: + await Self.runFixedRate( + configuration: configuration, + operation: operation, + onError: onError + ) + } + self.clearIfCurrent(generation: currentGeneration) + } + } + + public func stop() async { + self.generation &+= 1 + + let task = self.task + self.task = nil + + task?.cancel() + await task?.value + } + + public var isRunning: Bool { + self.task != nil + } + + private func clearIfCurrent(generation: UInt64) { + guard self.generation == generation else { + return + } + + self.task = nil + } + + private static func runFixedDelay(configuration: Configuration, operation: @escaping Operation, onError: @escaping ErrorHandler) async { + if !configuration.runImmediately { + guard await sleep(interval: configuration.interval, tolerance: configuration.tolerance) else { + return + } + } + + while !Task.isCancelled { + let shouldContinue = await runOperation( + operation: operation, + onError: onError, + errorPolicy: configuration.errorPolicy + ) + guard shouldContinue else { + return + } + + guard await sleep(interval: configuration.interval, tolerance: configuration.tolerance) else { + return + } + } + } + + private static func runFixedRate(configuration: Configuration, operation: @escaping Operation, onError: @escaping ErrorHandler) async { + var nextRun = ContinuousClock.now + if !configuration.runImmediately { + nextRun = nextRun.advanced(by: configuration.interval) + } + + while !Task.isCancelled { + let now = ContinuousClock.now + if nextRun > now { + let delay = now.duration(to: nextRun) + guard await sleep(interval: delay, tolerance: configuration.tolerance) else { + return + } + } + + let shouldContinue = await runOperation( + operation: operation, + onError: onError, + errorPolicy: configuration.errorPolicy + ) + guard shouldContinue else { + return + } + + let afterRun = ContinuousClock.now + repeat { + nextRun = nextRun.advanced(by: configuration.interval) + } while nextRun <= afterRun + } + } + + private static func runOperation(operation: @escaping Operation, onError: @escaping ErrorHandler, errorPolicy: ErrorPolicy) async -> Bool { + do { + try Task.checkCancellation() + try await operation() + return true + } catch is CancellationError { + return false + } catch { + await onError(error) + + switch errorPolicy { + case .keepRunning(let delay): + if let delay { + return await sleep(interval: delay, tolerance: nil) + } + return true + case .stop: + return false + } + } + } + + private static func sleep(interval: Duration, tolerance: Duration?) async -> Bool { + do { + try await Task.sleep(for: interval, tolerance: tolerance) + return true + } catch is CancellationError { + return false + } catch { + return false + } + } +} diff --git a/Tun/Punchnet/SDLAsyncTimerStream.swift b/Tun/Punchnet/Concurrency/SDLAsyncTimerStream.swift similarity index 100% rename from Tun/Punchnet/SDLAsyncTimerStream.swift rename to Tun/Punchnet/Concurrency/SDLAsyncTimerStream.swift