Fix schedule(after:interval:leeway:) disposal

CHANGELOG.md

@@ -1,6 +1,8 @@
 # master
 *Please add new entries at the top.*
 
+1. Fixed `schedule(after:interval:leeway:)` being cancelled when the returned `Disposable` is not retained.
+
 # 3.1.0-rc.1
 1. Fixed a scenario of downstream interruptions being dropped. (#577, kudos to @andersio)
 

Sources/Scheduler.swift

@@ -193,8 +193,14 @@ public final class QueueScheduler: DateScheduler {
 
 	public let queue: DispatchQueue
 
+	private let lock: Lock
+
+	private var timers: [DispatchSourceTimer]
+
 	internal init(internalQueue: DispatchQueue) {
 		queue = internalQueue
+		lock = Lock.make()
+		timers = []
 	}
 
 	/// Initializes a scheduler that will target the given queue with its
@@ -340,8 +346,16 @@ public final class QueueScheduler: DateScheduler {
 		timer.setEventHandler(handler: action)
 		timer.resume()
 
+		lock.lock()
+		timers.append(timer)
+		lock.unlock()
+
 		return AnyDisposable {
 			timer.cancel()
+
+			self.lock.lock()
+			self.timers = self.timers.filter { !$0.isEqual(timer) }
+			self.lock.unlock()
 		}
 	}
 }

Tests/ReactiveSwiftTests/SchedulerSpec.swift

@@ -218,6 +218,34 @@ class SchedulerSpec: QuickSpec {
 					scheduler.queue.resume()
 					expect{count}.toEventually(equal(timesToRun))
 				}
+
+				it("should repeatedly run actions after a given date when the disposable is not retained") {
+					let disposable = SerialDisposable()
+
+					var count = 0
+					let timesToIncrement = 3
+
+					// Start two repeating timers, dispose the first, and ensure only the second runs.
+
+					disposable.inner = scheduler.schedule(after: Date(), interval: .milliseconds(10), leeway: .seconds(0)) {
+						expect(false).to(equal(true), description: "timer not cancelled on disposal")
+					}
+
+					scheduler.schedule(after: Date(), interval: .milliseconds(10), leeway: .seconds(0)) {
+						expect(Thread.isMainThread) == false
+
+						if count <= timesToIncrement {
+							count += 1
+						}
+					}
+
+					disposable.dispose()
+
+					expect(count) == 0
+
+					scheduler.queue.resume()
+					expect{count}.toEventually(equal(timesToIncrement))
+				}
 			}
 		}