Add WallClockScheduler

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
 

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

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<String>()
+        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<String>()
+        let queue = DispatchQueue(label: #function)
+        queue.setSpecific(key: key, value: #function)
+        let scheduler = WallClockScheduler(queue: queue)
+        let subject = PassthroughSubject<Int, Never>()
+        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
+    )
+}