Repository: broadwaylamb/OpenCombine
Pull request #114 (2019-12-02 14:42:49)

Implement Delay
(picking up #55)

codecov[bot]: # [Codecov](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/114?src=pr&el=h1) Report
> Merging [#114](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/114?src=pr&el=desc) into [master](https://codecov.io/gh/broadwaylamb/OpenCombine/commit/d6b4fb41150ce6680ca1f4165974797f80c6eab6?src=pr&el=desc) will **increase** coverage by `0.11%`.
> The diff coverage is `100%`.

[![Impacted file tree graph](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/114/graphs/tree.svg?width=650&token=XMi8wfbUeP&height=150&src=pr)](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/114?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #114      +/-   ##
==========================================
+ Coverage   97.94%   98.06%   +0.11%     
==========================================
  Files          66       67       +1     
  Lines        3311     3404      +93     
==========================================
+ Hits         3243     3338      +95     
+ Misses         68       66       -2
```


| [Impacted Files](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/114?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...rces/OpenCombine/Publishers/Publishers.Delay.swift](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/114/diff?src=pr&el=tree#diff-U291cmNlcy9PcGVuQ29tYmluZS9QdWJsaXNoZXJzL1B1Ymxpc2hlcnMuRGVsYXkuc3dpZnQ=) | `100% <100%> (ø)` | |
| [...rces/OpenCombine/Publishers/Result.Publisher.swift](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/114/diff?src=pr&el=tree#diff-U291cmNlcy9PcGVuQ29tYmluZS9QdWJsaXNoZXJzL1Jlc3VsdC5QdWJsaXNoZXIuc3dpZnQ=) | `100% <0%> (ø)` | :arrow_up: |
| [Sources/OpenCombine/ObservableObject.swift](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/114/diff?src=pr&el=tree#diff-U291cmNlcy9PcGVuQ29tYmluZS9PYnNlcnZhYmxlT2JqZWN0LnN3aWZ0) | `75% <0%> (+15%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/114?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/114?src=pr&el=footer). Last update [d6b4fb4...a0d741a](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/114?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

OpenCombineBot: 
<!--
  0 failure: 
  0 warning: 
  
  1 markdown notices
  DangerID: danger-id-Danger;
-->



LGTM
<p align="right">
  Generated by :no_entry_sign: <a href="https://danger.systems/swift">Danger Swift</a> against 92c5a80da6e79851c4c6b7639ce3c09638c81479
</p>


Patch:
diff --git a/RemainingCombineInterface.swift b/RemainingCombineInterface.swift
index e9c4f5a..820ad2a 100644
--- a/RemainingCombineInterface.swift
+++ b/RemainingCombineInterface.swift
@@ -1795,58 +1795,6 @@ extension Publisher {
     public func tryCatch<P>(_ handler: @escaping (Self.Failure) throws -> P) -> Publishers.TryCatch<Self, P> where P : Publisher, Self.Output == P.Output
 }
 
-extension Publishers {
-
-    /// A publisher that delays delivery of elements and completion to the downstream receiver.
-    public struct Delay<Upstream, Context> : Publisher where Upstream : Publisher, Context : Scheduler {
-
-        /// The kind of values published by this publisher.
-        public typealias Output = Upstream.Output
-
-        /// The kind of errors this publisher might publish.
-        ///
-        /// Use `Never` if this `Publisher` does not publish errors.
-        public typealias Failure = Upstream.Failure
-
-        /// The publisher that this publisher receives elements from.
-        public let upstream: Upstream
-
-        /// The amount of time to delay.
-        public let interval: Context.SchedulerTimeType.Stride
-
-        /// The allowed tolerance in firing delayed events.
-        public let tolerance: Context.SchedulerTimeType.Stride
-
-        /// The scheduler to deliver the delayed events.
-        public let scheduler: Context
-
-        public let options: Context.SchedulerOptions?
-
-        public init(upstream: Upstream, interval: Context.SchedulerTimeType.Stride, tolerance: Context.SchedulerTimeType.Stride, scheduler: Context, options: Context.SchedulerOptions? = nil)
-
-        /// This function is called to attach the specified `Subscriber` to this `Publisher` by `subscribe(_:)`
-        ///
-        /// - SeeAlso: `subscribe(_:)`
-        /// - Parameters:
-        ///     - subscriber: The subscriber to attach to this `Publisher`.
-        ///                   once attached it can begin to receive values.
-        public func receive<S>(subscriber: S) where S : Subscriber, Upstream.Failure == S.Failure, Upstream.Output == S.Input
-    }
-}
-
-extension Publisher {
-
-    /// Delays delivery of all output to the downstream receiver by a specified amount of time on a particular scheduler.
-    ///
-    /// The delay affects the delivery of elements and completion, but not of the original subscription.
-    /// - Parameters:
-    ///   - interval: The amount of time to delay.
-    ///   - tolerance: The allowed tolerance in firing delayed events.
-    ///   - scheduler: The scheduler to deliver the delayed events.
-    /// - Returns: A publisher that delays delivery of elements and completion to the downstream receiver.
-    public func delay<S>(for interval: S.SchedulerTimeType.Stride, tolerance: S.SchedulerTimeType.Stride? = nil, scheduler: S, options: S.SchedulerOptions? = nil) -> Publishers.Delay<Self, S> where S : Scheduler
-}
-
 extension Just {
 
     public func prepend(_ elements: Output...) -> Publishers.Sequence<[Output], Just<Output>.Failure>
diff --git a/Sources/OpenCombine/Publishers/Publishers.Delay.swift b/Sources/OpenCombine/Publishers/Publishers.Delay.swift
new file mode 100644
index 0000000..e5f5254
--- /dev/null
+++ b/Sources/OpenCombine/Publishers/Publishers.Delay.swift
@@ -0,0 +1,227 @@
+//
+//  Publishers.Delay.swift
+//  OpenCombine
+//
+//  Created by Евгений Богомолов on 07/09/2019.
+//
+
+import COpenCombineHelpers
+
+extension Publisher {
+
+    /// Delays delivery of all output to the downstream receiver by a specified amount
+    /// of time on a particular scheduler.
+    ///
+    /// The delay affects the delivery of elements and completion, but not of the original
+    /// subscription.
+    ///
+    /// - Parameters:
+    ///   - interval: The amount of time to delay.
+    ///   - tolerance: The allowed tolerance in firing delayed events.
+    ///   - scheduler: The scheduler to deliver the delayed events.
+    /// - Returns: A publisher that delays delivery of elements and completion to
+    ///   the downstream receiver.
+    public func delay<Context: Scheduler>(
+        for interval: Context.SchedulerTimeType.Stride,
+        tolerance: Context.SchedulerTimeType.Stride? = nil,
+        scheduler: Context,
+        options: Context.SchedulerOptions? = nil
+    ) -> Publishers.Delay<Self, Context> {
+        return .init(upstream: self,
+                     interval: interval,
+                     tolerance: tolerance ?? scheduler.minimumTolerance,
+                     scheduler: scheduler,
+                     options: options)
+    }
+}
+
+extension Publishers {
+
+    /// A publisher that delays delivery of elements and completion
+    /// to the downstream receiver.
+    public struct Delay<Upstream: Publisher, Context: Scheduler>: Publisher {
+
+        public typealias Output = Upstream.Output
+
+        public typealias Failure = Upstream.Failure
+
+        /// The publisher that this publisher receives elements from.
+        public let upstream: Upstream
+
+        /// The amount of time to delay.
+        public let interval: Context.SchedulerTimeType.Stride
+
+        /// The allowed tolerance in firing delayed events.
+        public let tolerance: Context.SchedulerTimeType.Stride
+
+        /// The scheduler to deliver the delayed events.
+        public let scheduler: Context
+
+        public let options: Context.SchedulerOptions?
+
+        public init(upstream: Upstream,
+                    interval: Context.SchedulerTimeType.Stride,
+                    tolerance: Context.SchedulerTimeType.Stride,
+                    scheduler: Context,
+                    options: Context.SchedulerOptions? = nil)
+        {
+            self.upstream = upstream
+            self.interval = interval
+            self.tolerance = tolerance
+            self.scheduler = scheduler
+            self.options = options
+        }
+
+        public func receive<Downstream: Subscriber>(subscriber: Downstream)
+            where Upstream.Failure == Downstream.Failure,
+                  Upstream.Output == Downstream.Input
+        {
+            upstream.subscribe(Inner(self, downstream: subscriber))
+        }
+    }
+}
+
+extension Publishers.Delay {
+    private final class Inner<Downstream: Subscriber>
+        : Subscriber,
+          Subscription
+        where Downstream.Input == Upstream.Output, Downstream.Failure == Upstream.Failure
+    {
+        // NOTE: This class has been audited for thread safety
+
+        typealias Input = Upstream.Output
+        typealias Failure = Upstream.Failure
+
+        fileprivate typealias Delay = Publishers.Delay<Upstream, Context>
+
+        private enum State {
+            case ready(Delay, Downstream)
+            case subscribed(Delay, Downstream, Subscription)
+            case terminal
+        }
+
+        private let lock = UnfairLock.allocate()
+        private var state: State
+        private let downstreamLock = UnfairLock.allocate()
+
+        fileprivate init(_ publisher: Delay, downstream: Downstream) {
+            state = .ready(publisher, downstream)
+        }
+
+        deinit {
+            lock.deallocate()
+            downstreamLock.deallocate()
+        }
+
+        private func schedule(_ delay: Delay,
+                              immediate: Bool,
+                              work: @escaping () -> Void) {
+            if immediate {
+                delay.scheduler.schedule(options: delay.options, work)
+                return
+            }
+            delay
+                .scheduler
+                .schedule(after: delay.scheduler.now.advanced(by: delay.interval),
+                          tolerance: delay.tolerance,
+                          options: delay.options,
+                          work)
+        }
+
+        func receive(subscription: Subscription) {
+            lock.lock()
+            guard case let .ready(delay, downstream) = state else {
+                lock.unlock()
+                subscription.cancel()
+                return
+            }
+            state = .subscribed(delay, downstream, subscription)
+            lock.unlock()
+            schedule(delay, immediate: true) { [weak self] in
+                self?.scheduledReceive(subscription: subscription)
+            }
+        }
+
+        private func scheduledReceive(subscription: Subscription) {
+            lock.lock()
+            guard case let .subscribed(_, downstream, _) = state else {
+                lock.unlock()
+                return
+            }
+            lock.unlock()
+            downstreamLock.lock()
+            downstream.receive(subscription: self)
+            downstreamLock.unlock()
+        }
+
+        func receive(_ input: Upstream.Output) -> Subscribers.Demand {
+            lock.lock()
+            guard case let .subscribed(delay, downstream, _) = state else {
+                lock.unlock()
+                return .none
+            }
+            lock.unlock()
+            schedule(delay, immediate: false) { [weak self] in
+                self?.scheduledReceive(input, downstream: downstream)
+            }
+            return .none
+        }
+
+        private func scheduledReceive(_ input: Upstream.Output, downstream: Downstream) {
+            downstreamLock.lock()
+            let newDemand = downstream.receive(input)
+            downstreamLock.unlock()
+            guard newDemand > 0 else {
+                return
+            }
+            lock.lock()
+            guard case let .subscribed(_, _, subscription) = state else {
+                lock.unlock()
+                return
+            }
+            lock.unlock()
+            subscription.request(newDemand)
+        }
+
+        func receive(completion: Subscribers.Completion<Failure>) {
+            lock.lock()
+            guard case let .subscribed(delay, downstream, _) = state else {
+                lock.unlock()
+                return
+            }
+            state = .terminal
+            lock.unlock()
+            schedule(delay, immediate: false) { [weak self] in
+                self?.scheduledReceive(completion: completion, downstream: downstream)
+            }
+        }
+
+        private func scheduledReceive(completion: Subscribers.Completion<Failure>,
+                                      downstream: Downstream) {
+            downstreamLock.lock()
+            downstream.receive(completion: completion)
+            downstreamLock.unlock()
+        }
+
+        func request(_ demand: Subscribers.Demand) {
+            lock.lock()
+            guard case let .subscribed(_, _, subscription) = state else {
+                lock.unlock()
+                return
+            }
+            lock.unlock()
+            subscription.request(demand)
+        }
+
+        func cancel() {
+            lock.lock()
+            guard case let .subscribed(_, _, subscription) = state else {
+                lock.unlock()
+                return
+            }
+            state = .terminal
+            lock.unlock()
+            subscription.cancel()
+        }
+    }
+}


Test patch:
diff --git a/Tests/OpenCombineTests/Helpers/CustomPublisher.swift b/Tests/OpenCombineTests/Helpers/CustomPublisher.swift
index 551143a..e78bebf 100644
--- a/Tests/OpenCombineTests/Helpers/CustomPublisher.swift
+++ b/Tests/OpenCombineTests/Helpers/CustomPublisher.swift
@@ -34,7 +34,7 @@ import OpenCombine
 typealias CustomPublisher = CustomPublisherBase<Int, TestingError>
 
 @available(macOS 10.15, iOS 13.0, *)
-class CustomPublisherBase<Output, Failure: Error>: Publisher {
+class CustomPublisherBase<Output, Failure: Error>: Publisher, Cancellable {
 
     private(set) var subscriber: AnySubscriber<Output, Failure>?
     private(set) var erasedSubscriber: Any?
@@ -67,6 +67,11 @@ class CustomPublisherBase<Output, Failure: Error>: Publisher {
     func send(completion: Subscribers.Completion<Failure>) {
         subscriber?.receive(completion: completion)
     }
+
+    func cancel() {
+        subscriber = nil
+        erasedSubscriber = nil
+    }
 }
 
 @available(macOS 10.15, iOS 13.0, *)
diff --git a/Tests/OpenCombineTests/Helpers/PriorityQueue.swift b/Tests/OpenCombineTests/Helpers/PriorityQueue.swift
new file mode 100644
index 0000000..152eb08
--- /dev/null
+++ b/Tests/OpenCombineTests/Helpers/PriorityQueue.swift
@@ -0,0 +1,111 @@
+//
+//  PriorityQueue.swift
+//  
+//
+//  Created by Sergej Jaskiewicz on 02.12.2019.
+//
+
+/// A priproty queue based on binary max-heap.
+struct PriorityQueue<Element> {
+    private var storage: [Element]
+    private let areInIncreasingOrder: (Element, Element) -> Bool
+
+    init(_ areInIncreasingOrder: @escaping (Element, Element) -> Bool) {
+        self.storage = []
+        self.areInIncreasingOrder = areInIncreasingOrder
+    }
+
+    mutating func insert(_ element: Element) {
+        storage.append(element)
+        var newElementIndex = storage.endIndex - 1
+        while let parent = self.parent(of: newElementIndex),
+              areInIncreasingOrder(storage[parent], storage[newElementIndex]) {
+            storage.swapAt(newElementIndex, parent)
+            newElementIndex = parent
+        }
+    }
+
+    func max() -> Element? {
+        return storage.first
+    }
+
+    @discardableResult
+    mutating func extractMax() -> Element? {
+        guard let max = storage.first else { return nil }
+        storage[0] = storage[storage.endIndex - 1]
+        storage.removeLast()
+        maxHeapify(0)
+        return max
+    }
+
+    var count: Int {
+        return storage.count
+    }
+
+    var isEmpty: Bool {
+        return storage.isEmpty
+    }
+
+    private func leftChild(of index: Int) -> Int? {
+        assert(index >= 0)
+        let childIndex = 2 * index + 1
+        return childIndex < storage.endIndex ? childIndex : nil
+    }
+
+    private func rightChild(of index: Int) -> Int? {
+        assert(index >= 0)
+        let childIndex = 2 * index + 2
+        return childIndex < storage.endIndex ? childIndex : nil
+    }
+
+    private func parent(of index: Int) -> Int? {
+        assert(index >= 0)
+        if index == 0 { return nil }
+        return (index - 1) / 2
+    }
+
+    private mutating func maxHeapify(_ root: Int) {
+        var root = root
+        var largest = root
+        while true {
+            assert(largest == root)
+            if let left = leftChild(of: root),
+               areInIncreasingOrder(storage[root], storage[left]) {
+                largest = left
+            }
+            if let right = rightChild(of: root),
+               areInIncreasingOrder(storage[largest], storage[right]) {
+                largest = right
+            }
+            if largest == root {
+                break
+            }
+            storage.swapAt(root, largest)
+            root = largest
+        }
+    }
+}
+
+extension PriorityQueue where Element: Comparable {
+    init() {
+        self.init(<)
+    }
+}
+
+extension PriorityQueue: Sequence {
+    struct Iterator: IteratorProtocol {
+        private var queue: PriorityQueue
+
+        fileprivate init(_ queue: PriorityQueue) {
+            self.queue = queue
+        }
+
+        mutating func next() -> Element? {
+            return queue.extractMax()
+        }
+    }
+
+    func makeIterator() -> Iterator {
+        return Iterator(self)
+    }
+}
diff --git a/Tests/OpenCombineTests/Helpers/TestReflection.swift b/Tests/OpenCombineTests/Helpers/TestReflection.swift
index 63d3823..5230a00 100644
--- a/Tests/OpenCombineTests/Helpers/TestReflection.swift
+++ b/Tests/OpenCombineTests/Helpers/TestReflection.swift
@@ -79,9 +79,9 @@ internal func testReflection<Output, Failure: Error, Operator: Publisher>(
     line: UInt = #line,
     parentInput: Output.Type,
     parentFailure: Failure.Type,
-    description expectedDescription: String,
+    description expectedDescription: String?,
     customMirror customMirrorPredicate: ((Mirror) -> Bool)?,
-    playgroundDescription: String,
+    playgroundDescription: String?,
     _ makeOperator: (CustomConnectablePublisherBase<Output, Failure>) -> Operator
 ) throws {
     let publisher = CustomConnectablePublisherBase<Output, Failure>(subscription: nil)
@@ -97,16 +97,18 @@ internal func testReflection<Output, Failure: Error, Operator: Publisher>(
                    file: file,
                    line: line)
 
-    let customMirror =
-        try XCTUnwrap((erasedSubscriber as? CustomReflectable)?.customMirror,
-                      file: file,
-                      line: line)
-
     if let customMirrorPredicate = customMirrorPredicate {
+        let customMirror =
+            try XCTUnwrap((erasedSubscriber as? CustomReflectable)?.customMirror,
+                          file: file,
+                          line: line)
         XCTAssert(customMirrorPredicate(customMirror),
                   "customMirror doesn't satisfy the predicate",
                   file: file,
                   line: line)
+    } else {
+        XCTAssertFalse(erasedSubscriber is CustomReflectable,
+                       "subscriber shouldn't conform to CustomReflectable")
     }
 
     XCTAssertFalse(erasedSubscriber is CustomDebugStringConvertible,
diff --git a/Tests/OpenCombineTests/Helpers/VirtualTimeScheduler.swift b/Tests/OpenCombineTests/Helpers/VirtualTimeScheduler.swift
new file mode 100644
index 0000000..c200278
--- /dev/null
+++ b/Tests/OpenCombineTests/Helpers/VirtualTimeScheduler.swift
@@ -0,0 +1,264 @@
+//
+//  VirtualTimeScheduler.swift
+//  OpenCombineTests
+//
+//  Created by Евгений Богомолов on 14/09/2019.
+//
+
+#if OPENCOMBINE_COMPATIBILITY_TEST
+import Combine
+#else
+import OpenCombine
+#endif
+
+@available(macOS 10.15, iOS 13.0, *)
+final class VirtualTimeScheduler: Scheduler {
+
+    struct SchedulerTimeType: Strideable, Comparable, Hashable {
+
+        struct Stride: ExpressibleByFloatLiteral,
+                       Comparable,
+                       SignedNumeric,
+                       SchedulerTimeIntervalConvertible
+        {
+            var magnitude: Int64
+
+            fileprivate init(magnitude: Int64) {
+                self.magnitude = magnitude
+            }
+
+            init(integerLiteral value: Int) {
+                self = .seconds(value)
+            }
+
+            init(floatLiteral value: Double) {
+                self = .seconds(value)
+            }
+
+            init?<Source: BinaryInteger>(exactly source: Source) {
+                guard let magnitude = Int64(exactly: source) else {
+                    return nil
+                }
+                self.init(magnitude: magnitude)
+            }
+
+            static func < (lhs: Stride, rhs: Stride) -> Bool {
+                return lhs.magnitude < rhs.magnitude
+            }
+
+            static func * (lhs: Stride, rhs: Stride) -> Stride {
+                return Stride(magnitude: lhs.magnitude * rhs.magnitude)
+            }
+
+            static func + (lhs: Stride, rhs: Stride) -> Stride {
+                return Stride(magnitude: lhs.magnitude + rhs.magnitude)
+            }
+
+            static func - (lhs: Stride, rhs: Stride) -> Stride {
+                return Stride(magnitude: lhs.magnitude - rhs.magnitude)
+            }
+
+            static func -= (lhs: inout Stride, rhs: Stride) {
+                lhs.magnitude -= rhs.magnitude
+            }
+
+            static func *= (lhs: inout Stride, rhs: Stride) {
+                lhs.magnitude *= rhs.magnitude
+            }
+
+            static func += (lhs: inout Stride, rhs: Stride) {
+                lhs.magnitude += rhs.magnitude
+            }
+
+            static func seconds(_ value: Int) -> Stride {
+                return Stride(magnitude: Int64(value) * 1_000_000_000)
+            }
+
+            static func seconds(_ value: Double) -> Stride {
+                return Stride(magnitude: Int64(value * 1_000_000_000))
+            }
+
+            static func milliseconds(_ value: Int) -> Stride {
+                return Stride(magnitude: Int64(value) * 1_000_000)
+            }
+
+            static func microseconds(_ value: Int) -> Stride {
+                return Stride(magnitude: Int64(value) * 1_000)
+            }
+
+            static func nanoseconds(_ value: Int) -> Stride {
+                return Stride(magnitude: Int64(value))
+            }
+        }
+
+        /// Time in virtual nanoseconds
+        let time: UInt64
+
+        init(nanoseconds time: UInt64) {
+            self.time = time
+        }
+
+        static func == (lhs: SchedulerTimeType, rhs: SchedulerTimeType) -> Bool {
+            return lhs.time == rhs.time
+        }
+
+        static func < (lhs: SchedulerTimeType, rhs: SchedulerTimeType) -> Bool {
+            return lhs.time < rhs.time
+        }
+
+        func hash(into hasher: inout Hasher) {
+            hasher.combine(time)
+        }
+
+        func distance(to other: SchedulerTimeType) -> Stride {
+            if self > other {
+                return Stride(magnitude: -Int64(time - other.time))
+            } else {
+                return Stride(magnitude: Int64(other.time - time))
+            }
+        }
+
+        func advanced(by stride: Stride) -> SchedulerTimeType {
+            return stride.magnitude < 0
+                ? SchedulerTimeType(nanoseconds: time - UInt64(-stride.magnitude))
+                : SchedulerTimeType(nanoseconds: time + UInt64(stride.magnitude))
+        }
+
+        static func + (lhs: SchedulerTimeType, rhs: Stride) -> SchedulerTimeType {
+            return lhs.advanced(by: rhs)
+        }
+    }
+
+    struct SchedulerOptions {}
+
+    private struct EnqueuedAction: Comparable {
+        let time: SchedulerTimeType
+        let action: () -> Void
+
+        init(time: SchedulerTimeType, action: @escaping () -> Void) {
+            self.time = time
+            self.action = action
+        }
+
+        static func < (lhs: EnqueuedAction, rhs: EnqueuedAction) -> Bool {
+            return lhs.time > rhs.time
+        }
+
+        static func == (lhs: EnqueuedAction, rhs: EnqueuedAction) -> Bool {
+            return lhs.time == rhs.time
+        }
+    }
+
+    private final class CancellableToken: Cancellable {
+
+        private(set) var isCancelled = false
+
+        func cancel() {
+            isCancelled = true
+        }
+    }
+
+    enum Event: Equatable, CustomStringConvertible {
+        case now
+        case minimumTolerance
+        case schedule
+        case scheduleAfterDate(SchedulerTimeType,
+                               tolerance: SchedulerTimeType.Stride)
+        case scheduleAfterDateWithInterval(SchedulerTimeType,
+                                           interval: SchedulerTimeType.Stride,
+                                           tolerance: SchedulerTimeType.Stride)
+
+        var description: String {
+            switch self {
+            case .now:
+                return ".now"
+            case .minimumTolerance:
+                return ".minimumTolerance"
+            case .schedule:
+                return ".schedule"
+            case let .scheduleAfterDate(date, tolerance):
+                return """
+                .scheduleAfterDate(.init(nanoseconds: \(date.time)), \
+                tolerance: .nanoseconds(\(tolerance.magnitude)))
+                """
+            case let .scheduleAfterDateWithInterval(date, interval, tolerance):
+                return """
+                .scheduleAfterDateWithInterval(.init(nanoseconds: \(date.time)), \
+                interval: .nanoseconds(\(interval.magnitude)), \
+                tolerance: .nanoseconds(\(tolerance.magnitude)))
+                """
+            }
+        }
+    }
+
+    private(set) var history = [Event]()
+
+    private var _now = SchedulerTimeType(nanoseconds: 0)
+
+    private var workQueue = PriorityQueue<EnqueuedAction>()
+
+    var scheduledDates: [SchedulerTimeType] {
+        return workQueue.map { $0.time }
+    }
+
+    var now: SchedulerTimeType {
+        history.append(.now)
+        return _now
+    }
+
+    var minimumTolerance: SchedulerTimeType.Stride {
+        history.append(.minimumTolerance)
+        return 0
+    }
+
+    func schedule(options: SchedulerOptions?, _ action: @escaping () -> Void) {
+        history.append(.schedule)
+        workQueue.insert(.init(time: _now, action: action))
+    }
+
+    func schedule(after date: SchedulerTimeType,
+                  tolerance: SchedulerTimeType.Stride,
+                  options: SchedulerOptions?,
+                  _ action: @escaping () -> Void) {
+        history.append(.scheduleAfterDate(date, tolerance: tolerance))
+        workQueue.insert(.init(time: date, action: action))
+    }
+
+    func schedule(after date: SchedulerTimeType,
+                  interval: SchedulerTimeType.Stride,
+                  tolerance: SchedulerTimeType.Stride,
+                  options: SchedulerOptions?,
+                  _ action: @escaping () -> Void) -> Cancellable {
+        history.append(.scheduleAfterDateWithInterval(date,
+                                                      interval: interval,
+                                                      tolerance: tolerance))
+        let cancellableToken = CancellableToken()
+        repeatedlyExecute(after: date,
+                          interval: interval,
+                          cancellableToken: cancellableToken,
+                          action: action)
+        return cancellableToken
+    }
+
+    private func repeatedlyExecute(after date: SchedulerTimeType,
+                                   interval: SchedulerTimeType.Stride,
+                                   cancellableToken: CancellableToken,
+                                   action: @escaping () -> Void) {
+        let enqueuedAction = EnqueuedAction(time: date) { [unowned self] in
+            if cancellableToken.isCancelled { return }
+            action()
+            self.repeatedlyExecute(after: date + interval,
+                                   interval: interval,
+                                   cancellableToken: cancellableToken,
+                                   action: action)
+        }
+        workQueue.insert(enqueuedAction)
+    }
+
+    func executeScheduledActions() {
+        while let enqueuedAction = workQueue.extractMax() {
+            _now = enqueuedAction.time
+            enqueuedAction.action()
+        }
+    }
+}
diff --git a/Tests/OpenCombineTests/PublisherTests/DeferredTests.swift b/Tests/OpenCombineTests/PublisherTests/DeferredTests.swift
index 734ac07..6c5d917 100644
--- a/Tests/OpenCombineTests/PublisherTests/DeferredTests.swift
+++ b/Tests/OpenCombineTests/PublisherTests/DeferredTests.swift
@@ -7,7 +7,6 @@
 
 import XCTest
 
-//import Combine
 #if OPENCOMBINE_COMPATIBILITY_TEST
 import Combine
 #else
diff --git a/Tests/OpenCombineTests/PublisherTests/DelayTests.swift b/Tests/OpenCombineTests/PublisherTests/DelayTests.swift
new file mode 100644
index 0000000..cdc1831
--- /dev/null
+++ b/Tests/OpenCombineTests/PublisherTests/DelayTests.swift
@@ -0,0 +1,443 @@
+//
+//  DelayTests.swift
+//  OpenCombineTests
+//
+//  Created by Евгений Богомолов on 08/09/2019.
+//
+
+import XCTest
+
+#if OPENCOMBINE_COMPATIBILITY_TEST
+import Combine
+#else
+import OpenCombine
+#endif
+
+@available(macOS 10.15, iOS 13.0, *)
+final class DelayTests: XCTestCase {
+
+    // Delay's Inner doesn't conform to CustomStringConvertible, so we can't compare
+    // subscriptions using their descriptions
+    let delaySubscription: StringSubscription = {
+        let tracking = TrackingSubscriber()
+        let scheduler = VirtualTimeScheduler()
+        CustomPublisher(subscription: CustomSubscription())
+            .delay(for: 0, scheduler: scheduler)
+            .subscribe(tracking)
+        scheduler.executeScheduledActions()
+        return tracking.subscriptions.first.map(StringSubscription.subscription)
+            ?? "Delay"
+    }()
+
+    func testBasicBehavior() {
+        let scheduler = VirtualTimeScheduler()
+        let helper = OperatorTestHelper(publisherType: CustomPublisher.self,
+                                        initialDemand: .max(100),
+                                        receiveValueDemand: .max(12)) {
+            $0.delay(for: .nanoseconds(200),
+                     tolerance: .nanoseconds(5),
+                     scheduler: scheduler,
+                     options: VirtualTimeScheduler.SchedulerOptions())
+        }
+        XCTAssertNotNil(helper.publisher.subscriber)
+
+        XCTAssertEqual(helper.tracking.history, [])
+        XCTAssertEqual(helper.subscription.history, [])
+        XCTAssertEqual(scheduler.history, [.schedule])
+
+        scheduler.executeScheduledActions()
+
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription)])
+        XCTAssertEqual(helper.subscription.history, [.requested(.max(100))])
+        XCTAssertEqual(scheduler.history, [.schedule])
+
+        XCTAssertEqual(helper.publisher.send(1), .none)
+        XCTAssertEqual(helper.publisher.send(2), .none)
+        XCTAssertEqual(helper.publisher.send(3), .none)
+
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription)])
+        XCTAssertEqual(helper.subscription.history, [.requested(.max(100))])
+        XCTAssertEqual(scheduler.scheduledDates, [.init(nanoseconds: 200),
+                                                  .init(nanoseconds: 200),
+                                                  .init(nanoseconds: 200)])
+
+        XCTAssertEqual(scheduler.history,
+                       [.schedule,
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 200),
+                                           tolerance: .nanoseconds(5)),
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 200),
+                                           tolerance: .nanoseconds(5)),
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 200),
+                                           tolerance: .nanoseconds(5))])
+
+        scheduler.executeScheduledActions()
+
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription),
+                                                 .value(1),
+                                                 .value(3),
+                                                 .value(2)])
+
+        XCTAssertEqual(helper.subscription.history, [.requested(.max(100)),
+                                                     .requested(.max(12)),
+                                                     .requested(.max(12)),
+                                                     .requested(.max(12))])
+
+        XCTAssertEqual(scheduler.history,
+                       [.schedule,
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 200),
+                                           tolerance: .nanoseconds(5)),
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 200),
+                                           tolerance: .nanoseconds(5)),
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 200),
+                                           tolerance: .nanoseconds(5))])
+
+        helper.publisher.send(completion: .failure(.oops))
+        helper.publisher.send(completion: .finished)
+        XCTAssertEqual(helper.publisher.send(4), .none)
+
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription),
+                                                 .value(1),
+                                                 .value(3),
+                                                 .value(2)])
+        XCTAssertEqual(helper.subscription.history, [.requested(.max(100)),
+                                                     .requested(.max(12)),
+                                                     .requested(.max(12)),
+                                                     .requested(.max(12))])
+        XCTAssertEqual(scheduler.scheduledDates, [.init(nanoseconds: 400)])
+        XCTAssertEqual(scheduler.history,
+                       [.schedule,
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 200),
+                                           tolerance: .nanoseconds(5)),
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 200),
+                                           tolerance: .nanoseconds(5)),
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 200),
+                                           tolerance: .nanoseconds(5)),
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 400),
+                                           tolerance: .nanoseconds(5))])
+
+        scheduler.executeScheduledActions()
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription),
+                                                 .value(1),
+                                                 .value(3),
+                                                 .value(2),
+                                                 .completion(.failure(.oops))])
+        XCTAssertEqual(helper.subscription.history, [.requested(.max(100)),
+                                                     .requested(.max(12)),
+                                                     .requested(.max(12)),
+                                                     .requested(.max(12))])
+        XCTAssertEqual(scheduler.history,
+                       [.schedule,
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 200),
+                                           tolerance: .nanoseconds(5)),
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 200),
+                                           tolerance: .nanoseconds(5)),
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 200),
+                                           tolerance: .nanoseconds(5)),
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 400),
+                                           tolerance: .nanoseconds(5))])
+        XCTAssertEqual(scheduler.now, .init(nanoseconds: 400))
+    }
+
+    func testRequest() throws {
+        let scheduler = VirtualTimeScheduler()
+        let helper = OperatorTestHelper(publisherType: CustomPublisher.self,
+                                        initialDemand: nil,
+                                        receiveValueDemand: .none) {
+            $0.delay(for: .nanoseconds(200),
+                     tolerance: .nanoseconds(5),
+                     scheduler: scheduler,
+                     options: VirtualTimeScheduler.SchedulerOptions())
+        }
+        scheduler.executeScheduledActions()
+
+        XCTAssertEqual(helper.subscription.history, [])
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription)])
+
+        try XCTUnwrap(helper.downstreamSubscription).request(.max(10))
+        try XCTUnwrap(helper.downstreamSubscription).request(.max(4))
+        try XCTUnwrap(helper.downstreamSubscription).request(.max(5))
+        try XCTUnwrap(helper.downstreamSubscription).request(.none)
+        XCTAssertEqual(helper.publisher.send(2000), .none)
+
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription)])
+        XCTAssertEqual(helper.subscription.history, [.requested(.max(10)),
+                                                     .requested(.max(4)),
+                                                     .requested(.max(5)),
+                                                     .requested(.none)])
+
+        scheduler.executeScheduledActions()
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription),
+                                                 .value(2000)])
+        XCTAssertEqual(helper.subscription.history, [.requested(.max(10)),
+                                                     .requested(.max(4)),
+                                                     .requested(.max(5)),
+                                                     .requested(.none)])
+    }
+
+    func testCancelAlreadyCancelled() throws {
+        let scheduler = VirtualTimeScheduler()
+        let helper = OperatorTestHelper(publisherType: CustomPublisher.self,
+                                        initialDemand: .unlimited,
+                                        receiveValueDemand: .none) {
+            $0.delay(for: .nanoseconds(200),
+                     tolerance: .nanoseconds(5),
+                     scheduler: scheduler,
+                     options: VirtualTimeScheduler.SchedulerOptions())
+        }
+
+        scheduler.executeScheduledActions()
+        XCTAssertEqual(helper.subscription.history, [.requested(.unlimited)])
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription)])
+
+        try XCTUnwrap(helper.downstreamSubscription).cancel()
+        try XCTUnwrap(helper.downstreamSubscription).request(.max(42))
+        try XCTUnwrap(helper.downstreamSubscription).cancel()
+
+        XCTAssertEqual(helper.subscription.history, [.requested(.unlimited), .cancelled])
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription)])
+        XCTAssertEqual(scheduler.history, [.schedule])
+
+        XCTAssertEqual(helper.publisher.send(0), .none)
+        helper.publisher.send(completion: .finished)
+
+        XCTAssertEqual(helper.subscription.history, [.requested(.unlimited), .cancelled])
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription)])
+        XCTAssertEqual(scheduler.history, [.schedule])
+    }
+
+    func testReceiveCompletionImmediatelyAfterSubscription() {
+        let scheduler = VirtualTimeScheduler()
+        let helper = OperatorTestHelper(publisherType: CustomPublisher.self,
+                                        initialDemand: .unlimited,
+                                        receiveValueDemand: .none) {
+            $0.delay(for: .nanoseconds(123),
+                     tolerance: .nanoseconds(5),
+                     scheduler: scheduler,
+                     options: VirtualTimeScheduler.SchedulerOptions())
+        }
+
+        helper.publisher.send(completion: .failure(.oops))
+
+        XCTAssertEqual(helper.tracking.history, [])
+        XCTAssertEqual(helper.subscription.history, [])
+        XCTAssertEqual(scheduler.history,
+                       [.schedule,
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 123),
+                                           tolerance: .nanoseconds(5))])
+
+        scheduler.executeScheduledActions()
+
+        XCTAssertEqual(helper.tracking.history, [.completion(.failure(.oops))])
+        XCTAssertEqual(helper.subscription.history, [])
+    }
+
+    func testReceiveCompletionImmediatelyAfterValue() {
+        let scheduler = VirtualTimeScheduler()
+        let helper = OperatorTestHelper(publisherType: CustomPublisher.self,
+                                        initialDemand: .unlimited,
+                                        receiveValueDemand: .max(418)) {
+            $0.delay(for: .nanoseconds(123),
+                     tolerance: .nanoseconds(5),
+                     scheduler: scheduler,
+                     options: VirtualTimeScheduler.SchedulerOptions())
+        }
+        XCTAssertEqual(helper.publisher.send(-1), .none)
+        scheduler.executeScheduledActions()
+
+        XCTAssertEqual(helper.publisher.send(1000), .none)
+        helper.publisher.send(completion: .finished)
+
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription),
+                                                 .value(-1)])
+        XCTAssertEqual(helper.subscription.history, [.requested(.unlimited),
+                                                     .requested(.max(418))])
+        XCTAssertEqual(scheduler.history,
+                       [.schedule,
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 123),
+                                           tolerance: .nanoseconds(5)),
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 246),
+                                           tolerance: .nanoseconds(5)),
+                        .now,
+                        .scheduleAfterDate(.init(nanoseconds: 246),
+                                           tolerance: .nanoseconds(5))])
+
+        scheduler.executeScheduledActions()
+
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription),
+                                                 .value(-1),
+                                                 .value(1000),
+                                                 .completion(.finished)])
+        XCTAssertEqual(helper.subscription.history, [.requested(.unlimited),
+                                                     .requested(.max(418))])
+    }
+
+    func testCrashesWhenReceivingInputRecursively() {
+        let helper = OperatorTestHelper(publisherType: CustomPublisher.self,
+                                        initialDemand: .unlimited,
+                                        receiveValueDemand: .max(418)) {
+            $0.delay(for: .nanoseconds(123), scheduler: ImmediateScheduler.shared)
+        }
+
+        helper.tracking.onValue = { _ in
+            _ = helper.publisher.send(-1)
+        }
+
+        assertCrashes {
+            _ = helper.publisher.send(0)
+        }
+    }
+
+    func testReceiveCompletionRecursively() {
+        let helper = OperatorTestHelper(publisherType: CustomPublisher.self,
+                                        initialDemand: .unlimited,
+                                        receiveValueDemand: .max(418)) {
+            $0.delay(for: .nanoseconds(123), scheduler: ImmediateScheduler.shared)
+        }
+        helper.tracking.onFinish = {
+            helper.publisher.send(completion: .finished)
+        }
+        helper.publisher.send(completion: .finished)
+    }
+
+    func testWeakCaptureWhenSchedulingSubscription() {
+        let scheduler = VirtualTimeScheduler()
+        var subscription: Subscription?
+        var subscriberReleased = false
+        do {
+            let publisher = CustomPublisher(subscription: CustomSubscription())
+            let delay = publisher.delay(for: 0.35, scheduler: scheduler)
+            let tracking = TrackingSubscriber(receiveSubscription: { subscription = $0 },
+                                              onDeinit: { subscriberReleased = true })
+            delay.subscribe(tracking)
+            XCTAssertEqual(tracking.history, [])
+            XCTAssertEqual(scheduler.history, [.minimumTolerance, .schedule])
+            publisher.cancel()
+        }
+        scheduler.executeScheduledActions()
+        XCTAssertNil(subscription)
+        XCTAssertTrue(subscriberReleased)
+    }
+
+    func testWeakCaptureWhenSchedulingValue() {
+        let scheduler = VirtualTimeScheduler()
+        var value: Int?
+        var subscriberReleased = false
+        do {
+            let publisher = CustomPublisher(subscription: CustomSubscription())
+            let delay = publisher.delay(for: 0.35, scheduler: scheduler)
+            let tracking = TrackingSubscriber(receiveValue: { value = $0; return .none },
+                                              onDeinit: { subscriberReleased = true })
+            delay.subscribe(tracking)
+            scheduler.executeScheduledActions()
+            XCTAssertEqual(tracking.history, [.subscription(delaySubscription)])
+            XCTAssertEqual(publisher.send(42), .none)
+            XCTAssertEqual(tracking.history, [.subscription(delaySubscription)])
+            XCTAssertEqual(scheduler.history,
+                           [.minimumTolerance,
+                            .schedule,
+                            .now,
+                            .scheduleAfterDate(.init(nanoseconds: 350000000),
+                                               tolerance: 0)])
+            tracking.cancel()
+            publisher.cancel()
+        }
+        scheduler.executeScheduledActions()
+        XCTAssertNil(value)
+        XCTAssertTrue(subscriberReleased)
+    }
+
+    func testWeakCaptureWhenSchedulingCompletion() {
+        let scheduler = VirtualTimeScheduler()
+        var completion: Subscribers.Completion<TestingError>?
+        var subscriberReleased = false
+        do {
+            let publisher = CustomPublisher(subscription: CustomSubscription())
+            let delay = publisher.delay(for: 0.35, scheduler: scheduler)
+            let tracking = TrackingSubscriber(receiveCompletion: { completion = $0 },
+                                              onDeinit: { subscriberReleased = true })
+            delay.subscribe(tracking)
+            scheduler.executeScheduledActions()
+            XCTAssertEqual(tracking.history, [.subscription(delaySubscription)])
+            publisher.send(completion: .finished)
+            XCTAssertEqual(tracking.history, [.subscription(delaySubscription)])
+            XCTAssertEqual(scheduler.history,
+                           [.minimumTolerance,
+                            .schedule,
+                            .now,
+                            .scheduleAfterDate(.init(nanoseconds: 350000000),
+                                               tolerance: 0)])
+            tracking.cancel()
+            publisher.cancel()
+        }
+        scheduler.executeScheduledActions()
+        XCTAssertNil(completion)
+        XCTAssertTrue(subscriberReleased)
+    }
+
+    func testDelayReceiveSubscriptionTwice() throws {
+        try testReceiveSubscriptionTwice {
+            $0.delay(for: 0.35, scheduler: ImmediateScheduler.shared)
+        }
+    }
+
+    func testDelayReceiveValueBeforeSubscription() {
+        testReceiveValueBeforeSubscription(value: 213,
+                                           expected: .history([], demand: .none)) {
+            $0.delay(for: 0.35, scheduler: ImmediateScheduler.shared)
+        }
+    }
+
+    func testDelayReceiveCompletionBeforeSubscription()  {
+        testReceiveCompletionBeforeSubscription(inputType: Int.self,
+                                                expected: .history([])) {
+            $0.delay(for: 0.35, scheduler: ImmediateScheduler.shared)
+        }
+    }
+
+    func testDelayRequestBeforeSubscription() {
+        testRequestBeforeSubscription(inputType: Int.self, shouldCrash: false) {
+            $0.delay(for: 0.35, scheduler: ImmediateScheduler.shared)
+        }
+    }
+
+    func testDelayCancelBeforeSubscription() {
+        testCancelBeforeSubscription(inputType: Int.self, shouldCrash: false) {
+            $0.delay(for: 0.35, scheduler: ImmediateScheduler.shared)
+        }
+    }
+
+    func testDelayReflection() throws {
+        /// Delay's Inner doesn't customize its reflection
+        try testReflection(parentInput: Int.self,
+                           parentFailure: Error.self,
+                           description: nil,
+                           customMirror: nil,
+                           playgroundDescription: nil) {
+            $0.delay(for: 42, scheduler: ImmediateScheduler.shared)
+        }
+    }
+
+    func testDelayLifecycle() throws {
+        try testLifecycle(sendValue: 31,
+                          cancellingSubscriptionReleasesSubscriber: true) {
+            $0.delay(for: 42, scheduler: ImmediateScheduler.shared)
+        }
+    }
+}
diff --git a/Tests/OpenCombineTests/VirtualTimeSchedulerTests.swift b/Tests/OpenCombineTests/VirtualTimeSchedulerTests.swift
new file mode 100644
index 0000000..e60284d
--- /dev/null
+++ b/Tests/OpenCombineTests/VirtualTimeSchedulerTests.swift
@@ -0,0 +1,104 @@
+//
+//  VirtualTimeSchedulerTests.swift
+//  OpenCombineTests
+//
+//  Created by Евгений Богомолов on 14/09/2019.
+//
+
+import XCTest
+
+@available(macOS 10.15, iOS 13.0, *)
+final class VirtualTimeSchedulerTests: XCTestCase {
+
+    func testOrder() {
+        var history = [Int]()
+        let scheduler = VirtualTimeScheduler()
+        scheduler.schedule(after: scheduler.now + .nanoseconds(10)) {
+            history.append(5)
+        }
+        scheduler.schedule {
+            history.append(1)
+        }
+        scheduler.schedule(after: scheduler.now + .nanoseconds(5)) {
+            history.append(3)
+            scheduler.schedule(after: scheduler.now + .nanoseconds(2)) {
+                history.append(4)
+            }
+        }
+        scheduler.schedule {
+            history.append(2)
+        }
+
+        XCTAssertEqual(scheduler.now, .init(nanoseconds: 0))
+        XCTAssertEqual(scheduler.scheduledDates, [.init(nanoseconds: 0),
+                                                  .init(nanoseconds: 0),
+                                                  .init(nanoseconds: 5),
+                                                  .init(nanoseconds: 10)])
+
+        scheduler.executeScheduledActions()
+        XCTAssertEqual(history, [1, 2, 3, 4, 5])
+        XCTAssertEqual(scheduler.now, .init(nanoseconds: 10))
+        XCTAssertEqual(scheduler.scheduledDates, [])
+
+        XCTAssertEqual(scheduler.history, [.now,
+                                           .minimumTolerance,
+                                           .scheduleAfterDate(.init(nanoseconds: 10),
+                                                              tolerance: 0),
+                                           .schedule,
+                                           .now,
+                                           .minimumTolerance,
+                                           .scheduleAfterDate(.init(nanoseconds: 5),
+                                                              tolerance: 0),
+                                           .schedule,
+                                           .now,
+                                           .now,
+                                           .minimumTolerance,
+                                           .scheduleAfterDate(.init(nanoseconds: 7),
+                                                              tolerance: 0),
+                                           .now])
+    }
+
+    func testRepeadedAction() {
+        let scheduler = VirtualTimeScheduler()
+        var history = [Int]()
+        let cancellable = scheduler.schedule(after: scheduler.now + .microseconds(2),
+                                             interval: .milliseconds(40)) {
+            history.append(Int(scheduler.now.time))
+        }
+        scheduler.schedule(after: scheduler.now + .milliseconds(300)) {
+            cancellable.cancel()
+        }
+        XCTAssertEqual(scheduler.scheduledDates, [.init(nanoseconds: 2_000),
+                                                  .init(nanoseconds: 300_000_000)])
+        scheduler.executeScheduledActions()
+
+        XCTAssertEqual(history, [2000,
+                                 40002000,
+                                 80002000,
+                                 120002000,
+                                 160002000,
+                                 200002000,
+                                 240002000,
+                                 280002000])
+        XCTAssertEqual(scheduler.now, .init(nanoseconds: 320002000))
+        XCTAssertEqual(scheduler.history,
+                       [.now,
+                        .minimumTolerance,
+                        .scheduleAfterDateWithInterval(.init(nanoseconds: 2000),
+                                                       interval: .nanoseconds(40000000),
+                                                       tolerance: 0),
+                        .now,
+                        .minimumTolerance,
+                        .scheduleAfterDate(.init(nanoseconds: 300000000),
+                                           tolerance: .nanoseconds(0)),
+                        .now,
+                        .now,
+                        .now,
+                        .now,
+                        .now,
+                        .now,
+                        .now,
+                        .now,
+                        .now])
+    }
+}