Merge pull request #505 from ReactiveCocoa/anders/migrate-operators

Improve TransformerCore and move `take(first:)`.

Author: mdiep

Sources/Event.swift

@@ -180,7 +180,9 @@ extension Signal.Event: EventProtocol {
 }
 
 extension Signal.Event {
-	internal static func filter(_ isIncluded: @escaping (Value) -> Bool) -> (@escaping Signal<Value, Error>.Observer.Action) -> (Signal<Value, Error>.Event) -> Void {
+	internal typealias Transformation<U, E: Swift.Error> = (@escaping Signal<U, E>.Observer.Action) -> (Signal<Value, Error>.Event) -> Void
+
+	internal static func filter(_ isIncluded: @escaping (Value) -> Bool) -> Transformation<Value, Error> {
 		return { action in
 			return { event in
 				switch event {
@@ -202,7 +204,7 @@ extension Signal.Event {
 		}
 	}
 
-	internal static func filterMap<U>(_ transform: @escaping (Value) -> U?) -> (@escaping Signal<U, Error>.Observer.Action) -> (Signal<Value, Error>.Event) -> Void {
+	internal static func filterMap<U>(_ transform: @escaping (Value) -> U?) -> Transformation<U, Error> {
 		return { action in
 			return { event in
 				switch event {
@@ -224,7 +226,7 @@ extension Signal.Event {
 		}
 	}
 
-	internal static func map<U>(_ transform: @escaping (Value) -> U) -> (@escaping Signal<U, Error>.Observer.Action) -> (Signal<Value, Error>.Event) -> Void {
+	internal static func map<U>(_ transform: @escaping (Value) -> U) -> Transformation<U, Error> {
 		return { action in
 			return { event in
 				switch event {
@@ -244,7 +246,7 @@ extension Signal.Event {
 		}
 	}
 
-	internal static func mapError<E>(_ transform: @escaping (Error) -> E) -> (@escaping Signal<Value, E>.Observer.Action) -> (Signal<Value, Error>.Event) -> Void {
+	internal static func mapError<E>(_ transform: @escaping (Error) -> E) -> Transformation<Value, E> {
 		return { action in
 			return { event in
 				switch event {
@@ -263,4 +265,28 @@ extension Signal.Event {
 			}
 		}
 	}
+
+	internal static func take(first count: Int) -> Transformation<Value, Error> {
+		assert(count >= 1)
+
+		return { action in
+			var taken = 0
+
+			return { event in
+				guard let value = event.value else {
+					action(event)
+					return
+				}
+
+				if taken < count {
+					taken += 1
+					action(.value(value))
+				}
+
+				if taken == count {
+					action(.completed)
+				}
+			}
+		}
+	}
 }

Sources/Observer.swift

@@ -30,8 +30,27 @@ extension Signal {
 		/// - parameters:
 		///   - observer: The observer to transform.
 		///   - transform: The transform.
-		internal init<U, E: Swift.Error>(_ observer: Signal<U, E>.Observer, _ transform: @escaping (@escaping Signal<U, E>.Observer.Action) -> Action) {
-			self.action = transform(observer.action)
+		///   - disposable: The disposable to be disposed of when the `TransformerCore`
+		///                 yields any terminal event. If `observer` is a `Signal` input
+		///                 observer, this can be omitted.
+		internal init<U, E: Swift.Error>(
+			_ observer: Signal<U, E>.Observer,
+			_ transform: @escaping Event.Transformation<U, E>,
+			_ disposable: Disposable? = nil
+		) {
+			var hasDeliveredTerminalEvent = false
+
+			self.action = transform { event in
+				if !hasDeliveredTerminalEvent {
+					observer.action(event)
+
+					if event.isTerminating {
+						hasDeliveredTerminalEvent = true
+						disposable?.dispose()
+					}
+				}
+			}
+
 			self.wrapped = observer.interruptsOnDeinit ? observer : nil
 			self.interruptsOnDeinit = false
 		}

Sources/Signal.swift

@@ -537,9 +537,9 @@ extension Signal {
 	///                closure.
 	///
 	/// - returns: A signal that forwards events yielded by the action.
-	internal func flatMapEvent<U, E>(_ transform: @escaping (@escaping Signal<U, E>.Observer.Action) -> (Event) -> Void) -> Signal<U, E> {
+	internal func flatMapEvent<U, E>(_ transform: @escaping Event.Transformation<U, E>) -> Signal<U, E> {
 		return Signal<U, E> { observer in
-			return self.observe(.init(observer, transform))
+			return self.observe(Signal.Observer(observer, transform))
 		}
 	}
 
@@ -642,31 +642,8 @@ extension Signal {
 	/// - returns: A signal that will yield the first `count` values from `self`
 	public func take(first count: Int) -> Signal<Value, Error> {
 		precondition(count >= 0)
-
-		return Signal { observer in
-			if count == 0 {
-				observer.sendCompleted()
-				return nil
-			}
-
-			var taken = 0
-
-			return self.observe { event in
-				guard let value = event.value else {
-					observer.action(event)
-					return
-				}
-
-				if taken < count {
-					taken += 1
-					observer.send(value: value)
-				}
-
-				if taken == count {
-					observer.sendCompleted()
-				}
-			}
-		}
+		guard count >= 1 else { return .empty }
+		return flatMapEvent(Signal.Event.take(first: count))
 	}
 }
 

Sources/SignalProducer.swift

@@ -70,8 +70,8 @@ public struct SignalProducer<Value, Error: Swift.Error> {
 			let interruptHandle = AnyDisposable(observer.sendInterrupted)
 
 			return SignalProducerCore.Instance(signal: signal,
-			                                                observerDidSetup: observerDidSetup,
-			                                                interruptHandle: interruptHandle)
+			                                   observerDidSetup: observerDidSetup,
+			                                   interruptHandle: interruptHandle)
 		})
 	}
 
@@ -192,12 +192,17 @@ public struct SignalProducer<Value, Error: Swift.Error> {
 		self.init([ first, second ] + tail)
 	}
 
-	/// A producer for a Signal that will immediately complete without sending
-	/// any values.
+	/// A producer for a Signal that immediately completes without sending any values.
 	public static var empty: SignalProducer {
 		return SignalProducer(GeneratorCore { observer, _ in observer.sendCompleted() })
 	}
 
+	/// A producer for a Signal that immediately interrupts when started, without
+	/// sending any values.
+	internal static var interrupted: SignalProducer {
+		return SignalProducer(GeneratorCore { observer, _ in observer.sendInterrupted() })
+	}
+
 	/// A producer for a Signal that never sends any events to its observers.
 	public static var never: SignalProducer {
 		return self.init { observer, lifetime in
@@ -243,7 +248,17 @@ internal class SignalProducerCore<Value, Error: Swift.Error> {
 		fatalError()
 	}
 
-	func start(_ observer: Signal<Value, Error>.Observer) -> Disposable {
+	/// Start the producer with an observer created by the given generator.
+	///
+	/// The created observer **must** manaully dispose of the given upstream interrupt
+	/// handle iff it performs any event transformation that might result in a terminal
+	/// event.
+	///
+	/// - parameters:
+	///   - generator: The closure to generate an observer.
+	///
+	/// - returns: A disposable to interrupt the started producer instance.
+	func start(_ generator: (_ upstreamInterruptHandle: Disposable) -> Signal<Value, Error>.Observer) -> Disposable {
 		fatalError()
 	}
 
@@ -257,7 +272,7 @@ internal class SignalProducerCore<Value, Error: Swift.Error> {
 	///                closure.
 	///
 	/// - returns: A producer that forwards events yielded by the action.
-	internal func flatMapEvent<U, E>(_ transform: @escaping (@escaping Signal<U, E>.Observer.Action) -> (Signal<Value, Error>.Event) -> Void) -> SignalProducer<U, E> {
+	internal func flatMapEvent<U, E>(_ transform: @escaping Signal<Value, Error>.Event.Transformation<U, E>) -> SignalProducer<U, E> {
 		return SignalProducer<U, E>(TransformerCore(source: self, transform: transform))
 	}
 }
@@ -269,9 +284,9 @@ private final class SignalCore<Value, Error: Swift.Error>: SignalProducerCore<Va
 		self._make = action
 	}
 
-	override func start(_ observer: Signal<Value, Error>.Observer) -> Disposable {
+	override func start(_ generator: (Disposable) -> Signal<Value, Error>.Observer) -> Disposable {
 		let instance = makeInstance()
-		instance.signal.observe(observer)
+		instance.signal.observe(generator(instance.interruptHandle))
 		instance.observerDidSetup()
 		return instance.interruptHandle
 	}
@@ -301,18 +316,18 @@ private final class SignalCore<Value, Error: Swift.Error>: SignalProducerCore<Va
 /// - note: This core does not use `Signal` unless it is requested via `makeInstance()`.
 private final class TransformerCore<Value, Error: Swift.Error, SourceValue, SourceError: Swift.Error>: SignalProducerCore<Value, Error> {
 	private let source: SignalProducerCore<SourceValue, SourceError>
-	private let transform: (@escaping Signal<Value, Error>.Observer.Action) -> (Signal<SourceValue, SourceError>.Event) -> Void
+	private let transform: Signal<SourceValue, SourceError>.Event.Transformation<Value, Error>
 
-	init(source: SignalProducerCore<SourceValue, SourceError>, transform: @escaping (@escaping Signal<Value, Error>.Observer.Action) -> (Signal<SourceValue, SourceError>.Event) -> Void) {
+	init(source: SignalProducerCore<SourceValue, SourceError>, transform: @escaping Signal<SourceValue, SourceError>.Event.Transformation<Value, Error>) {
 		self.source = source
 		self.transform = transform
 	}
 
-	internal override func start(_ observer: Signal<Value, Error>.Observer) -> Disposable {
-		return source.start(.init(observer, transform))
+	internal override func start(_ generator: (Disposable) -> Signal<Value, Error>.Observer) -> Disposable {
+		return source.start { Signal.Observer(generator($0), transform, $0) }
 	}
 
-	internal override func flatMapEvent<U, E>(_ transform: @escaping (@escaping Signal<U, E>.Observer.Action) -> (Signal<Value, Error>.Event) -> Void) -> SignalProducer<U, E> {
+	internal override func flatMapEvent<U, E>(_ transform: @escaping Signal<Value, Error>.Event.Transformation<U, E>) -> SignalProducer<U, E> {
 		return SignalProducer<U, E>(TransformerCore<U, E, SourceValue, SourceError>(source: source) { [innerTransform = self.transform] action in
 			return innerTransform(transform(action))
 		})
@@ -321,12 +336,12 @@ private final class TransformerCore<Value, Error: Swift.Error, SourceValue, Sour
 	internal override func makeInstance() -> Instance {
 		let product = source.makeInstance()
 		let signal = Signal<Value, Error> { observer in
-			return product.signal.observe(.init(observer, transform))
+			return product.signal.observe(Signal.Observer(observer, transform))
 		}
 
 		return Instance(signal: signal,
-		                             observerDidSetup: product.observerDidSetup,
-		                             interruptHandle: product.interruptHandle)
+		                observerDidSetup: product.observerDidSetup,
+		                interruptHandle: product.interruptHandle)
 	}
 }
 
@@ -347,11 +362,11 @@ private final class GeneratorCore<Value, Error: Swift.Error>: SignalProducerCore
 		self.generator = generator
 	}
 
-	internal override func start(_ observer: Signal<Value, Error>.Observer) -> Disposable {
+	internal override func start(_ observerGenerator: (Disposable) -> Signal<Value, Error>.Observer) -> Disposable {
 		// Object allocation is a considerable overhead. So unless the core is configured
 		// to be disposable, we would reuse the already-disposed, shared `NopDisposable`.
 		let d: Disposable = isDisposable ? _SimpleDisposable() : NopDisposable.shared
-		generator(observer, d)
+		generator(observerGenerator(d), d)
 		return d
 	}
 
@@ -469,7 +484,7 @@ extension SignalProducer {
 	/// - returns: A disposable to interrupt the produced `Signal`.
 	@discardableResult
 	public func start(_ observer: Signal<Value, Error>.Observer = .init()) -> Disposable {
-		return core.start(observer)
+		return core.start { _ in observer }
 	}
 
 	/// Create a `Signal` from `self`, and observe the `Signal` for all events
@@ -864,7 +879,8 @@ extension SignalProducer {
 	/// - returns: A producer that, when started, will yield the first `count`
 	///            values from `self`.
 	public func take(first count: Int) -> SignalProducer<Value, Error> {
-		return lift { $0.take(first: count) }
+		guard count >= 1 else { return .interrupted }
+		return core.flatMapEvent(Signal.Event.take(first: count))
 	}
 
 	/// Yield an array of values when `self` completes.