From 06f8cc5f3fbd518c228ba06f686ed44944a8ae65 Mon Sep 17 00:00:00 2001 From: Anthony Drendel Date: Tue, 14 Jul 2026 18:08:54 +0200 Subject: [PATCH] Add WallClockScheduler --- README.md | 1 + .../WallClockScheduler.swift | 190 ++++++++++++++++++ .../WallClockSchedulerTests.swift | 154 ++++++++++++++ 3 files changed, 345 insertions(+) create mode 100644 Sources/CombineExtensions/WallClockScheduler.swift create mode 100644 Tests/CombineExtensionsTests/WallClockSchedulerTests.swift diff --git a/README.md b/README.md index 90e11f7..9bffc45 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,7 @@ A collection of useful extensions for Apple's [Combine framework](https://develo - [TestScheduler](https://github.com/shareup/combine-extensions/blob/main/Sources/CombineExtensions/TestScheduler.swift) - [UIScheduler](https://github.com/shareup/combine-extensions/blob/main/Sources/CombineExtensions/UIScheduler.swift) +- [WallClockScheduler](https://github.com/shareup/combine-extensions/blob/main/Sources/CombineExtensions/WallClockScheduler.swift) ### Thread-safe subscription management diff --git a/Sources/CombineExtensions/WallClockScheduler.swift b/Sources/CombineExtensions/WallClockScheduler.swift new file mode 100644 index 0000000..3189660 --- /dev/null +++ b/Sources/CombineExtensions/WallClockScheduler.swift @@ -0,0 +1,190 @@ +import Combine +import Dispatch +import Foundation + +/// A scheduler that measures time using the wall clock and executes work on a dispatch queue. +/// +/// Unlike `DispatchQueue`'s `Scheduler` conformance, scheduled time advances while the system +/// sleeps. Work whose deadline passes while the process cannot run is submitted to `queue` as +/// soon as the process can run again. +public struct WallClockScheduler: Scheduler, Sendable { + public struct SchedulerTimeType: Strideable, Codable, Hashable, Sendable { + public typealias Stride = RunLoop.SchedulerTimeType.Stride + + public let date: Date + + public init(_ date: Date) { + self.date = date + } + + public func distance(to other: Self) -> Stride { + Stride(other.date.timeIntervalSince(date)) + } + + public func advanced(by stride: Stride) -> Self { + Self(date.addingTimeInterval(stride.timeInterval)) + } + } + + public typealias SchedulerOptions = DispatchQueue.SchedulerOptions + + public let queue: DispatchQueue + + public init(queue: DispatchQueue) { + self.queue = queue + } + + public var now: SchedulerTimeType { + SchedulerTimeType(Date()) + } + + public var minimumTolerance: SchedulerTimeType.Stride { + .zero + } + + public func schedule( + options: SchedulerOptions?, + _ action: @escaping () -> Void + ) { + let options = options ?? SchedulerOptions() + scheduleOnQueue(options: options, action) + } + + public func schedule( + after date: SchedulerTimeType, + tolerance _: SchedulerTimeType.Stride, + options: SchedulerOptions?, + _ action: @escaping () -> Void + ) { + let options = options ?? SchedulerOptions() + options.group?.enter() + + queue.asyncAfter( + wallDeadline: date.dispatchWallTime, + qos: options.qos, + flags: options.flags + ) { + defer { options.group?.leave() } + action() + } + } + + public func schedule( + after date: SchedulerTimeType, + interval: SchedulerTimeType.Stride, + tolerance: SchedulerTimeType.Stride, + options: SchedulerOptions?, + _ action: @escaping () -> Void + ) -> Cancellable { + let options = options ?? SchedulerOptions() + return WallClockRepeatingTimer( + queue: queue, + deadline: date.dispatchWallTime, + interval: interval.dispatchTimeInterval(minimumNanoseconds: 1), + tolerance: tolerance.dispatchTimeInterval(minimumNanoseconds: 0), + options: options, + action: action + ) + } + + private func scheduleOnQueue( + options: SchedulerOptions, + _ action: @escaping () -> Void + ) { + options.group?.enter() + queue.async( + qos: options.qos, + flags: options.flags + ) { + defer { options.group?.leave() } + action() + } + } +} + +public extension WallClockScheduler.SchedulerTimeType { + static func < (lhs: Self, rhs: Self) -> Bool { + lhs.date < rhs.date + } +} + +private final class WallClockRepeatingTimer: Cancellable, @unchecked Sendable { + private let source: DispatchSourceTimer + + init( + queue: DispatchQueue, + deadline: DispatchWallTime, + interval: DispatchTimeInterval, + tolerance: DispatchTimeInterval, + options: WallClockScheduler.SchedulerOptions, + action: @escaping () -> Void + ) { + source = DispatchSource.makeTimerSource(queue: queue) + source.setEventHandler(qos: options.qos, flags: options.flags) { + options.group?.enter() + defer { options.group?.leave() } + action() + } + source.schedule( + wallDeadline: deadline, + repeating: interval, + leeway: tolerance + ) + source.activate() + } + + deinit { + cancel() + } + + func cancel() { + source.cancel() + } +} + +private extension WallClockScheduler.SchedulerTimeType { + var dispatchWallTime: DispatchWallTime { + guard date > Date() else { return .now() } + + let interval = date.timeIntervalSince1970 + guard interval.isFinite, + interval < TimeInterval(UInt64.max) / nanosecondsPerSecond + else { return .distantFuture } + + var seconds = Int(floor(interval)) + var nanoseconds = Int( + ((interval - TimeInterval(seconds)) * nanosecondsPerSecond).rounded(.up) + ) + + if nanoseconds >= Int(nanosecondsPerSecond) { + seconds += 1 + nanoseconds = 0 + } + + return DispatchWallTime( + timespec: timespec( + tv_sec: seconds, + tv_nsec: nanoseconds + ) + ) + } +} + +private extension WallClockScheduler.SchedulerTimeType.Stride { + func dispatchTimeInterval(minimumNanoseconds: Int) -> DispatchTimeInterval { + let rawNanoseconds = (timeInterval * nanosecondsPerSecond).rounded(.up) + let nanoseconds: Int + + if rawNanoseconds.isNaN || rawNanoseconds <= Double(minimumNanoseconds) { + nanoseconds = minimumNanoseconds + } else if !rawNanoseconds.isFinite || rawNanoseconds >= Double(Int.max) { + nanoseconds = Int.max + } else { + nanoseconds = Int(rawNanoseconds) + } + + return .nanoseconds(nanoseconds) + } +} + +private let nanosecondsPerSecond: Double = 1_000_000_000 diff --git a/Tests/CombineExtensionsTests/WallClockSchedulerTests.swift b/Tests/CombineExtensionsTests/WallClockSchedulerTests.swift new file mode 100644 index 0000000..b3e8dab --- /dev/null +++ b/Tests/CombineExtensionsTests/WallClockSchedulerTests.swift @@ -0,0 +1,154 @@ +import Combine +import CombineExtensions +import XCTest + +final class WallClockSchedulerTests: XCTestCase { + func testSchedulerTimeTypeUsesDateArithmetic() { + let date = Date(timeIntervalSinceReferenceDate: 1000) + let time = WallClockScheduler.SchedulerTimeType(date) + + let advanced = time.advanced(by: .milliseconds(250)) + + XCTAssertEqual(advanced.date, date.addingTimeInterval(0.25)) + XCTAssertEqual(time.distance(to: advanced).timeInterval, 0.25, accuracy: 0.000_001) + } + + func testNowUsesCurrentWallClockDate() { + let scheduler = makeScheduler() + let before = Date() + + let now = scheduler.now.date + + XCTAssertGreaterThanOrEqual(now, before) + XCTAssertLessThanOrEqual(now, Date()) + } + + func testImmediateActionRunsOnDispatchQueue() { + let key = DispatchSpecificKey() + let queue = DispatchQueue(label: #function) + queue.setSpecific(key: key, value: #function) + let scheduler = WallClockScheduler(queue: queue) + let executed = expectation(description: "Executed") + + scheduler.schedule { + XCTAssertEqual(DispatchQueue.getSpecific(key: key), #function) + executed.fulfill() + } + + wait(for: [executed], timeout: 1) + } + + func testFutureActionRunsAtWallClockDeadline() { + let scheduler = makeScheduler() + let executed = expectation(description: "Executed") + let deadline = Date().addingTimeInterval(0.01) + var executedAt: Date? + + scheduler.schedule( + after: .init(deadline), + tolerance: .zero, + options: nil + ) { + executedAt = Date() + executed.fulfill() + } + + wait(for: [executed], timeout: 1) + XCTAssertGreaterThanOrEqual(executedAt ?? .distantPast, deadline) + } + + func testPastActionRunsAtNextOpportunity() { + let scheduler = makeScheduler() + let executed = expectation(description: "Executed") + + scheduler.schedule( + after: .init(.distantPast), + tolerance: .zero, + options: nil + ) { + executed.fulfill() + } + + wait(for: [executed], timeout: 1) + } + + func testRepeatingActionCanBeCancelled() { + let scheduler = makeScheduler() + let executed = expectation(description: "Executed") + executed.expectedFulfillmentCount = 3 + var cancellable: Cancellable? + var executionCount = 0 + + cancellable = scheduler.schedule( + after: scheduler.now.advanced(by: .milliseconds(1)), + interval: .milliseconds(1), + tolerance: .zero, + options: nil + ) { + executionCount += 1 + executed.fulfill() + + if executionCount == 3 { + cancellable?.cancel() + } + } + + wait(for: [executed], timeout: 1) + XCTAssertEqual(executionCount, 3) + } + + func testAgainAtRepublishesOnWallClockSchedulerQueue() { + let key = DispatchSpecificKey() + let queue = DispatchQueue(label: #function) + queue.setSpecific(key: key, value: #function) + let scheduler = WallClockScheduler(queue: queue) + let subject = PassthroughSubject() + let republished = expectation(description: "Republished") + var values = [Int]() + + let subscription = subject + .againAt(scheduler: scheduler) + .sink( + receiveCompletion: { _ in }, + receiveValue: { value, timer in + XCTAssertEqual(DispatchQueue.getSpecific(key: key), #function) + values.append(value) + + if values.count == 1 { + let date = Date().addingTimeInterval(0.01) + let time = timer.time(at: date) + XCTAssertDatesEqual(time.date, date, accuracy: 0.001) + timer.republish(at: time) + } else { + republished.fulfill() + } + } + ) + defer { subscription.cancel() } + + subject.send(1) + + wait(for: [republished], timeout: 1) + XCTAssertEqual(values, [1, 1]) + } + + private func makeScheduler() -> WallClockScheduler { + WallClockScheduler(queue: DispatchQueue(label: #function)) + } +} + +private func XCTAssertDatesEqual( + _ lhs: Date, + _ rhs: Date, + accuracy: TimeInterval, + file: StaticString = #filePath, + line: UInt = #line +) { + XCTAssertEqual( + lhs.timeIntervalSinceReferenceDate, + rhs.timeIntervalSinceReferenceDate, + accuracy: accuracy, + file: file, + line: line + ) +}