Refactored & Bugfix: startWithSignal and the hot-to-cold lift.

Sources/Signal.swift

@@ -128,14 +128,19 @@ public final class Signal<Value, Error: Swift.Error> {
 	/// observer.
 	///
 	/// - note: The Signal will remain alive until a terminating event is sent
-	///         to the observer.
+	///         to the observer, or until it has no observer if it is not
+	///         retained.
+	///
+	/// - parameters:
+	///   - disposable: An optional disposable to associate with the signal, and
+	///                 to be disposed of when the signal terminates.
 	///
 	/// - returns: A tuple made of signal and observer.
-	public static func pipe() -> (Signal, Observer) {
+	public static func pipe(disposable: Disposable? = nil) -> (Signal, Observer) {
 		var observer: Observer!
 		let signal = self.init { innerObserver in
 			observer = innerObserver
-			return nil
+			return disposable
 		}
 
 		return (signal, observer)

Sources/SignalProducer.swift

@@ -172,36 +172,23 @@ public struct SignalProducer<Value, Error: Swift.Error> {
 	/// - parameters:
 	///   - setUp: A closure that accepts a `signal` and `interrupter`.
 	public func startWithSignal(_ setup: (_ signal: Signal<Value, Error>, _ interrupter: Disposable) -> Void) {
-		let (signal, observer) = Signal<Value, Error>.pipe()
-
 		// Disposes of the work associated with the SignalProducer and any
 		// upstream producers.
 		let producerDisposable = CompositeDisposable()
 
+		let (signal, observer) = Signal<Value, Error>.pipe(disposable: producerDisposable)
+
 		// Directly disposed of when `start()` or `startWithSignal()` is
 		// disposed.
-		let cancelDisposable = ActionDisposable {
-			observer.sendInterrupted()
-			producerDisposable.dispose()
-		}
+		let cancelDisposable = ActionDisposable(action: observer.sendInterrupted)
 
 		setup(signal, cancelDisposable)
 
 		if cancelDisposable.isDisposed {
 			return
 		}
 
-		let wrapperObserver: Signal<Value, Error>.Observer = Observer { event in
-			observer.action(event)
-
-			if event.isTerminating {
-				// Dispose only after notifying the Signal, so disposal
-				// logic is consistently the last thing to run.
-				producerDisposable.dispose()
-			}
-		}
-
-		startHandler(wrapperObserver, producerDisposable)
+		startHandler(observer, producerDisposable)
 	}
 }
 
@@ -445,7 +432,6 @@ extension SignalProducerProtocol {
 			}
 		}
 	}
-
 
 	/// Lift a binary Signal operator to operate upon a Signal and a
 	/// SignalProducer instead.
@@ -462,27 +448,9 @@ extension SignalProducerProtocol {
 	///            `SignalProducer`.
 	public func lift<U, F, V, G>(_ transform: @escaping (Signal<Value, Error>) -> (Signal<U, F>) -> Signal<V, G>) -> (Signal<U, F>) -> SignalProducer<V, G> {
 		return { otherSignal in
-			return SignalProducer { observer, outerDisposable in
-				let (wrapperSignal, otherSignalObserver) = Signal<U, F>.pipe()
-
-				// Avoid memory leak caused by the direct use of the given
-				// signal.
-				//
-				// See https://github.com/ReactiveCocoa/ReactiveCocoa/pull/2758
-				// for the details.
-				outerDisposable += ActionDisposable {
-					otherSignalObserver.sendInterrupted()
-				}
-				outerDisposable += otherSignal.observe(otherSignalObserver)
-
-				self.startWithSignal { signal, disposable in
-					outerDisposable += disposable
-					outerDisposable += transform(signal)(wrapperSignal).observe(observer)
-				}
-			}
+			return self.liftRight(transform)(SignalProducer(signal: otherSignal))
 		}
 	}
-
 
 	/// Map each value in the producer to a new value.
 	///

Tests/ReactiveSwiftTests/SignalProducerSpec.swift

@@ -337,7 +337,8 @@ class SignalProducerSpec: QuickSpec {
 					return
 				}
 
-				producer.startWithSignal { _, innerDisposable in
+				producer.startWithSignal { signal, innerDisposable in
+					signal.observe { _ in }
 					disposable = innerDisposable
 				}
 
@@ -432,7 +433,7 @@ class SignalProducerSpec: QuickSpec {
 					observer = incomingObserver
 				}
 
-				producer.startWithSignal { _ in }
+				producer.start()
 				expect(addedDisposable.isDisposed) == false
 
 				observer.sendCompleted()
@@ -448,12 +449,31 @@ class SignalProducerSpec: QuickSpec {
 					observer = incomingObserver
 				}
 
-				producer.startWithSignal { _ in }
+				producer.start()
 				expect(addedDisposable.isDisposed) == false
 
 				observer.send(error: .default)
 				expect(addedDisposable.isDisposed) == true
 			}
+
+			it("should dispose of the added disposable if the signal is unretained and unobserved upon exiting the scope") {
+				let addedDisposable = SimpleDisposable()
+
+				let producer = SignalProducer<Int, TestError> { _, disposable in
+					disposable += addedDisposable
+				}
+
+				var started = false
+				var disposed = false
+
+				producer
+					.on(started: { started = true }, disposed: { disposed = true })
+					.startWithSignal { _ in }
+
+				expect(started) == true
+				expect(disposed) == true
+				expect(addedDisposable.isDisposed) == true
+			}
 		}
 
 		describe("start") {
@@ -1970,6 +1990,7 @@ class SignalProducerSpec: QuickSpec {
 				producer
 					.observe(on: TestScheduler())
 					.startWithSignal { signal, innerDisposable in
+						signal.observe { _ in }
 						downstreamDisposable = innerDisposable
 					}
 

Tests/ReactiveSwiftTests/SignalSpec.swift

@@ -192,6 +192,16 @@ class SignalSpec: QuickSpec {
 				expect(completed) == true
 			}
 
+			it("should dispose the supplied disposable when the signal terminates") {
+				let disposable = SimpleDisposable()
+				let (signal, observer) = Signal<(), NoError>.pipe(disposable: disposable)
+
+				expect(disposable.isDisposed) == false
+
+				observer.sendCompleted()
+				expect(disposable.isDisposed) == true
+			}
+
 			context("memory") {
 				it("should not crash allocating memory with a few observers") {
 					let (signal, _) = Signal<Int, NoError>.pipe()