Optimized producer lifting.

Author: andersio

Sources/SignalProducer.swift

@@ -2,7 +2,7 @@ import Dispatch
 import Foundation
 import Result
 
-/// A SignalProducer creates Signals that can produce values of type `Value` 
+/// A SignalProducer creates Signals that can produce values of type `Value`
 /// and/or fail with errors of type `Error`. If no failure should be possible, 
 /// `NoError` can be specified for `Error`.
 ///
@@ -19,7 +19,7 @@ import Result
 public struct SignalProducer<Value, Error: Swift.Error> {
 	public typealias ProducedSignal = Signal<Value, Error>
 
-	private let startHandler: (Signal<Value, Error>.Observer, CompositeDisposable) -> Void
+	private let backing: Backing<Value, Error>
 
 	/// Initializes a `SignalProducer` that will emit the same events as the
 	/// given signal.
@@ -30,9 +30,7 @@ public struct SignalProducer<Value, Error: Swift.Error> {
 	/// - parameters:
 	///   - signal: A signal to observe after starting the producer.
 	public init<S: SignalProtocol>(signal: S) where S.Value == Value, S.Error == Error {
-		self.init { observer, disposable in
-			disposable += signal.observe(observer)
-		}
+		backing = .signal { (signal.signal, {}) }
 	}
 
 	/// Initializes a SignalProducer that will invoke the given closure once for
@@ -49,7 +47,23 @@ public struct SignalProducer<Value, Error: Swift.Error> {
 	/// - parameters:
 	///   - startHandler: A closure that accepts observer and a disposable.
 	public init(_ startHandler: @escaping (Signal<Value, Error>.Observer, CompositeDisposable) -> Void) {
-		self.startHandler = startHandler
+		backing = .defer(startHandler)
+	}
+
+	/// Create a SignalProducer that will invoke the given factory once for each
+	/// invocation of `start()`.
+	///
+	/// - warning: `Signal` should not escape the factory unles absolutely
+	///            necessary.
+	///
+	/// - note: When nesting factories, remember to call the upstream's starting
+	///         side effect.
+	///
+	/// - parameters:
+	///   - factory: A factory that produces a `Signal` and its associated
+	///              starting side effects.
+	internal init(_ factory: @escaping () -> (Signal<Value, Error>, () -> Void)) {
+		backing = .signal(factory)
 	}
 
 	/// Creates a producer for a `Signal` that will immediately send one value
@@ -162,6 +176,128 @@ public struct SignalProducer<Value, Error: Swift.Error> {
 		}
 	}
 
+	/// Lift an unary Signal operator to operate upon SignalProducers instead.
+	///
+	/// In other words, this will create a new `SignalProducer` which will apply
+	/// the given `Signal` operator to _every_ created `Signal`, just as if the
+	/// operator had been applied to each `Signal` yielded from `start()`.
+	///
+	/// - parameters:
+	///   - transform: An unary operator to lift.
+	///
+	/// - returns: A signal producer that applies signal's operator to every
+	///            created signal.
+	fileprivate func _lift<U, F>(_ transform: @escaping (Signal<Value, Error>) -> Signal<U, F>) -> SignalProducer<U, F> {
+		switch backing {
+		case let .defer(body):
+			return SignalProducer<U, F> {
+				let producerDisposable = CompositeDisposable()
+				let (signal, observer) = Signal<Value, Error>.pipe(disposable: producerDisposable)
+				return (transform(signal), { body(observer, producerDisposable) })
+			}
+
+		case let .signal(factory):
+			return SignalProducer<U, F> {
+				let (signal, start) = factory()
+				return (transform(signal), start)
+			}
+		}
+	}
+
+	/// Lift a binary Signal operator to operate upon SignalProducers.
+	///
+	/// The start order of the producer must be specified. When both producers are
+	/// synchronous this order can be important depending on the operator to
+	/// generate correct results.
+	///
+	/// - parameters:
+	///   - order: The start order of the producers.
+	///
+	/// - returns: A factory that creates a SignalProducer with the given operator
+	///            applied. `self` would be the LHS, and the factory input would
+	///            be the RHS.
+	fileprivate func _lift<U, F, V, G>(_ order: StartOrder, _ transform: @escaping (Signal<Value, Error>) -> (Signal<U, F>) -> Signal<V, G>) -> (SignalProducer<U, F>) -> SignalProducer<V, G> {
+		return { otherProducer in
+			switch (self.backing, otherProducer.backing) {
+			case (let .defer(leftBody), let .defer(rightBody)):
+				return SignalProducer<V, G> {
+					let leftDisposable = CompositeDisposable()
+					let rightDisposable = CompositeDisposable()
+					let (leftSignal, leftObserver) = Signal<Value, Error>.pipe(disposable: leftDisposable)
+					let (rightSignal, rightObserver) = Signal<U, F>.pipe(disposable: rightDisposable)
+
+					return (transform(leftSignal)(rightSignal), {
+						switch order {
+						case .rightLeft:
+							rightBody(rightObserver, rightDisposable)
+							leftBody(leftObserver, leftDisposable)
+
+						case .leftRight:
+							leftBody(leftObserver, leftDisposable)
+							rightBody(rightObserver, rightDisposable)
+						}
+					})
+				}
+
+			case (let .defer(leftBody), let .signal(rightFactory)):
+				return SignalProducer<V, G> {
+					let leftDisposable = CompositeDisposable()
+					let (leftSignal, leftObserver) = Signal<Value, Error>.pipe(disposable: leftDisposable)
+					let (rightSignal, rightStart) = rightFactory()
+
+					return (transform(leftSignal)(rightSignal), {
+						switch order {
+						case .rightLeft:
+							rightStart()
+							leftBody(leftObserver, leftDisposable)
+
+						case .leftRight:
+							leftBody(leftObserver, leftDisposable)
+							rightStart()
+						}
+					})
+				}
+
+			case (let .signal(leftFactory), let .defer(rightBody)):
+				return SignalProducer<V, G> {
+					let (leftSignal, leftStart) = leftFactory()
+					let rightDisposable = CompositeDisposable()
+					let (rightSignal, rightObserver) = Signal<U, F>.pipe(disposable: rightDisposable)
+
+					return (transform(leftSignal)(rightSignal), {
+						switch order {
+						case .rightLeft:
+							rightBody(rightObserver, rightDisposable)
+							leftStart()
+
+						case .leftRight:
+							leftStart()
+							rightBody(rightObserver, rightDisposable)
+						}
+					})
+				}
+
+			case (let .signal(leftFactory), let .signal(rightFactory)):
+				return SignalProducer<V, G> {
+					let (leftSignal, leftStart) = leftFactory()
+					let (rightSignal, rightStart) = rightFactory()
+
+					return (transform(leftSignal)(rightSignal), {
+						switch order {
+						case .rightLeft:
+							rightStart()
+							leftStart()
+
+						case .leftRight:
+							leftStart()
+							rightStart()
+						}
+					})
+				}
+			}
+		}
+	}
+
 	/// Create a Signal from the producer, pass it into the given closure,
 	/// then start sending events on the Signal when the closure has returned.
 	///
@@ -187,11 +323,47 @@ public struct SignalProducer<Value, Error: Swift.Error> {
 		if cancelDisposable.isDisposed {
 			return
 		}
-
-		startHandler(observer, producerDisposable)
+			
+		switch backing {
+		case let .defer(body):
+			body(observer, producerDisposable)
+
+		case let .signal(factory):
+			// The resource cleanup in this mode relies on the `Signal` lifetime
+			// semantics.
+			//
+			// `signal` would be disposed of when `observer` is detached from it,
+			// causing it to cease its self retainment. Generally, this would trigger
+			// the propagation of disposal from this end of the Signal graph back to
+			// the ultimate upstreams, unless any intermediate `Signal` is escaped.
+			let (signal, start) = factory()
+			producerDisposable += signal.observe(observer)
+			start()
+		}
 	}
 }
 
+/// Used to differentiate variants of SignalProducer start handlers.
+internal enum Backing<Value, Error: Swift.Error> {
+	/// `defer`: The closure is passed directly an event emitter, and a composite
+	///          disposable to aggregate resource cleanup logic.
+	case `defer`((_ emitter: Signal<Value, Error>.Observer, _ disposable: CompositeDisposable) -> Void)
+
+	/// `signal`: The closure produces a `Signal` that would be observed by the
+	///           final produced `Signal` in `startWithSignal`. Then the starting
+	///           side effect associated would be invoked.
+	case signal(() -> (signal: Signal<Value, Error>, start: () -> Void))
+}
+
+/// The start order of `SignalProducer`s in a binary lifted operator.
+internal enum StartOrder {
+	/// Start the left producer first.
+	case leftRight
+
+	/// Start the right producer first.
+	case rightLeft
+}
+
 public protocol SignalProducerProtocol {
 	/// The type of values being sent on the producer
 	associatedtype Value
@@ -216,6 +388,39 @@ extension SignalProducer: SignalProducerProtocol {
 	}
 }
 
+extension SignalProducerProtocol {
+	/// Lift an unary Signal operator to operate upon SignalProducers instead.
+	///
+	/// In other words, this will create a new `SignalProducer` which will apply
+	/// the given `Signal` operator to _every_ created `Signal`, just as if the
+	/// operator had been applied to each `Signal` yielded from `start()`.
+	///
+	/// - parameters:
+	///   - transform: An unary operator to lift.
+	///
+	/// - returns: A signal producer that applies signal's operator to every
+	///            created signal.
+	public func lift<U, F>(_ transform: @escaping (Signal<Value, Error>) -> Signal<U, F>) -> SignalProducer<U, F> {
+		return producer._lift(transform)
+	}
+
+	/// Right-associative lifting of a binary signal operator over producers.
+	/// That is, the argument producer will be started before the receiver. When
+	/// both producers are synchronous this order can be important depending on
+	/// the operator to generate correct results.
+	fileprivate func liftRight<U, F, V, G>(_ transform: @escaping (Signal<Value, Error>) -> (Signal<U, F>) -> Signal<V, G>) -> (SignalProducer<U, F>) -> SignalProducer<V, G> {
+		return producer._lift(.rightLeft, transform)
+	}
+
+	/// Left-associative lifting of a binary signal operator over producers.
+	/// That is, the receiver will be started before the argument producer. When
+	/// both producers are synchronous this order can be important depending on
+	/// the operator to generate correct results.
+	fileprivate func liftLeft<U, F, V, G>(_ transform: @escaping (Signal<Value, Error>) -> (Signal<U, F>) -> Signal<V, G>) -> (SignalProducer<U, F>) -> SignalProducer<V, G> {
+		return producer._lift(.leftRight, transform)
+	}
+}
+
 extension SignalProducerProtocol {
 	/// Create a Signal from the producer, then attach the given observer to
 	/// the `Signal` as an observer.
@@ -353,28 +558,6 @@ extension SignalProducerProtocol where Error == NoError {
 }
 
 extension SignalProducerProtocol {
-	/// Lift an unary Signal operator to operate upon SignalProducers instead.
-	///
-	/// In other words, this will create a new `SignalProducer` which will apply
-	/// the given `Signal` operator to _every_ created `Signal`, just as if the
-	/// operator had been applied to each `Signal` yielded from `start()`.
-	///
-	/// - parameters:
-	///   - transform: An unary operator to lift.
-	///
-	/// - returns: A signal producer that applies signal's operator to every
-	///            created signal.
-	public func lift<U, F>(_ transform: @escaping (Signal<Value, Error>) -> Signal<U, F>) -> SignalProducer<U, F> {
-		return SignalProducer { observer, outerDisposable in
-			self.startWithSignal { signal, innerDisposable in
-				outerDisposable += innerDisposable
-
-				transform(signal).observe(observer)
-			}
-		}
-	}
-	
-
 	/// Lift a binary Signal operator to operate upon SignalProducers instead.
 	///
 	/// In other words, this will create a new `SignalProducer` which will apply
@@ -393,46 +576,6 @@ extension SignalProducerProtocol {
 		return liftRight(transform)
 	}
 
-	/// Right-associative lifting of a binary signal operator over producers.
-	/// That is, the argument producer will be started before the receiver. When
-	/// both producers are synchronous this order can be important depending on
-	/// the operator to generate correct results.
-	private func liftRight<U, F, V, G>(_ transform: @escaping (Signal<Value, Error>) -> (Signal<U, F>) -> Signal<V, G>) -> (SignalProducer<U, F>) -> SignalProducer<V, G> {
-		return { otherProducer in
-			return SignalProducer { observer, outerDisposable in
-				self.startWithSignal { signal, disposable in
-					outerDisposable.add(disposable)
-
-					otherProducer.startWithSignal { otherSignal, otherDisposable in
-						outerDisposable += otherDisposable
-
-						transform(signal)(otherSignal).observe(observer)
-					}
-				}
-			}
-		}
-	}
-
-	/// Left-associative lifting of a binary signal operator over producers.
-	/// That is, the receiver will be started before the argument producer. When
-	/// both producers are synchronous this order can be important depending on
-	/// the operator to generate correct results.
-	private func liftLeft<U, F, V, G>(_ transform: @escaping (Signal<Value, Error>) -> (Signal<U, F>) -> Signal<V, G>) -> (SignalProducer<U, F>) -> SignalProducer<V, G> {
-		return { otherProducer in
-			return SignalProducer { observer, outerDisposable in
-				otherProducer.startWithSignal { otherSignal, otherDisposable in
-					outerDisposable += otherDisposable
-					
-					self.startWithSignal { signal, disposable in
-						outerDisposable.add(disposable)
-
-						transform(signal)(otherSignal).observe(observer)
-					}
-				}
-			}
-		}
-	}
-
 	/// Lift a binary Signal operator to operate upon a Signal and a
 	/// SignalProducer instead.
 	///