ObserveOn, LazyMap, Delay (#815)

Author: andersio

ReactiveSwift.xcodeproj/project.pbxproj

@@ -405,96 +405,20 @@ extension Signal.Event {
 	}
 
 	internal static func observe(on scheduler: Scheduler) -> Transformation<Value, Error> {
-		return { action, lifetime in
-			lifetime.observeEnded {
-				scheduler.schedule {
-					action(.interrupted)
-				}
-			}
-
-			return Signal.Observer { event in
-				scheduler.schedule {
-					if !lifetime.hasEnded {
-						action(event)
-					}
-				}
-			}
+		return { downstream, lifetime in
+			Operators.ObserveOn(downstream: downstream, downstreamLifetime: lifetime, target: scheduler)
 		}
 	}
 
 	internal static func lazyMap<U>(on scheduler: Scheduler, transform: @escaping (Value) -> U) -> Transformation<U, Error> {
-		return { action, lifetime in
-			let box = Atomic<Value?>(nil)
-			let completionDisposable = SerialDisposable()
-			let valueDisposable = SerialDisposable()
-
-			lifetime += valueDisposable
-			lifetime += completionDisposable
-
-			lifetime.observeEnded {
-				scheduler.schedule {
-					action(.interrupted)
-				}
-			}
-
-			return Signal.Observer { event in
-				switch event {
-				case let .value(value):
-					// Schedule only when there is no prior outstanding value.
-					if box.swap(value) == nil {
-						valueDisposable.inner = scheduler.schedule {
-							if let value = box.swap(nil) {
-								action(.value(transform(value)))
-							}
-						}
-					}
-
-				case .completed, .failed:
-					// Completion and failure should not discard the outstanding
-					// value.
-					completionDisposable.inner = scheduler.schedule {
-						action(event.map(transform))
-					}
-
-				case .interrupted:
-					// `interrupted` overrides any outstanding value and any
-					// scheduled completion/failure.
-					valueDisposable.dispose()
-					completionDisposable.dispose()
-					scheduler.schedule {
-						action(.interrupted)
-					}
-				}
-			}
+		return { downstream, lifetime in
+			Operators.LazyMap(downstream: downstream, downstreamLifetime: lifetime, target: scheduler, transform: transform)
 		}
 	}
 
 	internal static func delay(_ interval: TimeInterval, on scheduler: DateScheduler) -> Transformation<Value, Error> {
-		precondition(interval >= 0)
-
-		return { action, lifetime in
-			lifetime.observeEnded {
-				scheduler.schedule {
-					action(.interrupted)
-				}
-			}
-
-			return Signal.Observer { event in
-				switch event {
-				case .failed, .interrupted:
-					scheduler.schedule {
-						action(event)
-					}
-
-				case .value, .completed:
-					let date = scheduler.currentDate.addingTimeInterval(interval)
-					scheduler.schedule(after: date) {
-						if !lifetime.hasEnded {
-							action(event)
-						}
-					}
-				}
-			}
+		return { downstream, lifetime in
+			Operators.Delay(downstream: downstream, downstreamLifetime: lifetime, target: scheduler, interval: interval)
 		}
 	}
 

Sources/Event.swift

@@ -0,0 +1,44 @@
+import Foundation
+
+extension Operators {
+	internal final class Delay<Value, Error: Swift.Error>: UnaryAsyncOperator<Value, Value, Error> {
+		let interval: TimeInterval
+		let targetWithClock: DateScheduler
+
+		init(
+			downstream: Observer<Value, Error>,
+			downstreamLifetime: Lifetime,
+			target: DateScheduler,
+			interval: TimeInterval
+		) {
+			precondition(interval >= 0)
+
+			self.interval = interval
+			self.targetWithClock = target
+			super.init(downstream: downstream, downstreamLifetime: downstreamLifetime, target: target)
+		}
+
+		
+		override func receive(_ value: Value) {
+			guard isActive else { return }
+
+			targetWithClock.schedule(after: computeNextDate()) {
+				self.unscheduledSend(value)
+			}
+		}
+
+		override func terminate(_ termination: Termination<Error>) {
+			if case .completed = termination {
+				targetWithClock.schedule(after: computeNextDate()) {
+					super.terminate(.completed)
+				}
+			} else {
+				super.terminate(termination)
+			}
+		}
+
+		private func computeNextDate() -> Date {
+			targetWithClock.currentDate.addingTimeInterval(interval)
+		}
+	}
+}

Sources/Observers/Delay.swift

@@ -0,0 +1,43 @@
+extension Operators {
+	internal final class LazyMap<Value, NewValue, Error: Swift.Error>: UnaryAsyncOperator<Value, NewValue, Error> {
+		let transform: (Value) -> NewValue
+		let box = Atomic<Value?>(nil)
+	  let valueDisposable = SerialDisposable()
+
+		init(
+			downstream: Observer<NewValue, Error>,
+			downstreamLifetime: Lifetime,
+			target: Scheduler,
+			transform: @escaping (Value) -> NewValue
+		) {
+			self.transform = transform
+			super.init(downstream: downstream, downstreamLifetime: downstreamLifetime, target: target)
+
+			downstreamLifetime += valueDisposable
+		}
+
+		override func receive(_ value: Value) {
+			// Schedule only when there is no prior outstanding value.
+			if box.swap(value) == nil {
+				valueDisposable.inner = target.schedule {
+					if let value = self.box.swap(nil) {
+						self.unscheduledSend(self.transform(value))
+					}
+				}
+			}
+		}
+
+		override func terminate(_ termination: Termination<Error>) {
+			if case .interrupted = termination {
+				// `interrupted` immediately cancels any scheduled value.
+				//
+				// On the other hand, completion and failure does not cancel anything, and is scheduled to run after any
+				// scheduled value. `valueDisposable` will naturally be disposed by `downstreamLifetime` as soon as the
+				// downstream has processed the termination.
+				valueDisposable.dispose()
+			}
+
+			super.terminate(termination)
+		}
+	}
+}

Sources/Observers/LazyMap.swift

@@ -0,0 +1,10 @@
+extension Operators {
+	internal final class ObserveOn<Value, Error: Swift.Error>: UnaryAsyncOperator<Value, Value, Error> {
+		override func receive(_ value: Value) {
+			target.schedule {
+				guard !self.downstreamLifetime.hasEnded else { return }
+				self.unscheduledSend(value)
+			}
+		}
+	}
+}

Sources/Observers/ObserveOn.swift

@@ -0,0 +1,64 @@
+internal class UnaryAsyncOperator<InputValue, OutputValue, Error: Swift.Error>: Observer<InputValue, Error> {
+	let downstreamLifetime: Lifetime
+	let target: Scheduler
+
+	/// Whether or not the downstream observer can still receive values or termination.
+	///
+	/// - note: This is a thread-safe atomic read. So you can use it in any part of `receive(_:)` or `terminate(_:)` to
+	///         attempt to early out before expensive scheduling or computation.
+	var isActive: Bool { state.is(.active) }
+
+	// Direct access is discouraged for subclasses by keeping this private.
+	private let downstream: Observer<OutputValue, Error>
+	private let state: UnsafeAtomicState<AsyncOperatorState>
+
+	public init(
+		downstream: Observer<OutputValue, Error>,
+		downstreamLifetime: Lifetime,
+		target: Scheduler
+	) {
+		self.downstream = downstream
+		self.downstreamLifetime = downstreamLifetime
+		self.target = target
+		self.state = UnsafeAtomicState(.active)
+
+		super.init()
+
+		downstreamLifetime.observeEnded {
+			if self.state.tryTransition(from: .active, to: .terminated) {
+				target.schedule {
+						downstream.terminate(.interrupted)
+				}
+			}
+		}
+	}
+
+	deinit {
+		state.deinitialize()
+	}
+
+	open override func receive(_ value: InputValue) { fatalError() }
+
+	/// Send a value to the downstream without any implicit scheduling on `target`.
+	///
+	/// - important: Subclasses must invoke this only after having hopped onto the target scheduler.
+	final func unscheduledSend(_ value: OutputValue) {
+		downstream.receive(value)
+	}
+
+	open override func terminate(_ termination: Termination<Error>) {
+		// The atomic transition here must happen **after** we hop onto the target scheduler. This is to preserve the timing
+		// behaviour observed in previous versions of ReactiveSwift.
+
+		target.schedule {
+			if self.state.tryTransition(from: .active, to: .terminated) {
+				self.downstream.terminate(termination)
+			}
+		}
+	}
+}
+
+private enum AsyncOperatorState: Int32 {
+	case active
+	case terminated
+}