Fix a SignalProducer.lift issue which may leak intermediate signals. (#808)

* Failing test case

* Fix a `SignalProducer.lift` issue which may leak intermediate signals.

* Update CHANGELOG.

* Failing test cases batch 2.

* Fix also liftLeft and liftRight.

Author: andersio

CHANGELOG.md

@@ -1,6 +1,7 @@
 # master
 *Please add new entries at the top.*
 1. Add `ExpressibleByNilLiteral` constraint to `OptionalProtocol` (#805, kudos to @nkristek)
+2. Fixed a `SignalProducer.lift` issue which may leak intermediate signals. (#808) 
 
 # 6.4.0
 1. Bump min. deployment target to iOS 9 when using swift packages to silence Xcode 12 warnings. Update Quick & Nibmle to the latest version when using swift packages.

Sources/SignalProducer.swift

@@ -666,7 +666,7 @@ extension SignalProducer {
 		return SignalProducer<U, F> { observer, lifetime in
 			self.startWithSignal { signal, interrupter in
 				lifetime += interrupter
-				transform(signal).observe(observer)
+				lifetime += transform(signal).observe(observer)
 			}
 		}
 	}
@@ -679,15 +679,15 @@ extension SignalProducer {
 	/// - 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 liftLeft<U, F, V, G>(_ transform: @escaping (Signal<Value, Error>) -> (Signal<U, F>) -> Signal<V, G>) -> (SignalProducer<U, F>) -> SignalProducer<V, G> {
+	internal func liftLeft<U, F, V, G>(_ transform: @escaping (Signal<Value, Error>) -> (Signal<U, F>) -> Signal<V, G>) -> (SignalProducer<U, F>) -> SignalProducer<V, G> {
 		return { right in
 			return SignalProducer<V, G> { observer, lifetime in
 				right.startWithSignal { rightSignal, rightInterrupter in
 					lifetime += rightInterrupter
 
 					self.startWithSignal { leftSignal, leftInterrupter in
 						lifetime += leftInterrupter
-						transform(leftSignal)(rightSignal).observe(observer)
+						lifetime += transform(leftSignal)(rightSignal).observe(observer)
 					}
 				}
 			}
@@ -702,15 +702,15 @@ extension SignalProducer {
 	/// - 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 liftRight<U, F, V, G>(_ transform: @escaping (Signal<Value, Error>) -> (Signal<U, F>) -> Signal<V, G>) -> (SignalProducer<U, F>) -> SignalProducer<V, G> {
+	internal func liftRight<U, F, V, G>(_ transform: @escaping (Signal<Value, Error>) -> (Signal<U, F>) -> Signal<V, G>) -> (SignalProducer<U, F>) -> SignalProducer<V, G> {
 		return { right in
 			return SignalProducer<V, G> { observer, lifetime in
 				self.startWithSignal { leftSignal, leftInterrupter in
 					lifetime += leftInterrupter
 
 					right.startWithSignal { rightSignal, rightInterrupter in
 						lifetime += rightInterrupter
-						transform(leftSignal)(rightSignal).observe(observer)
+						lifetime += transform(leftSignal)(rightSignal).observe(observer)
 					}
 				}
 			}

Tests/ReactiveSwiftTests/SignalLifetimeSpec.swift

@@ -24,6 +24,7 @@ class SignalLifetimeSpec: QuickSpec {
 			it("should automatically interrupt if the input observer is not retained") {
 				let disposable = AnyDisposable()
 				var outerSignal: Signal<Never, Never>!
+				_ = outerSignal
 
 				func scope() {
 					let (signal, observer) = Signal<Never, Never>.pipe(disposable: disposable)
@@ -42,6 +43,7 @@ class SignalLifetimeSpec: QuickSpec {
 				let disposable = AnyDisposable()
 				var isInterrupted = false
 				var outerSignal: Signal<Never, Never>!
+				_ = outerSignal
 
 				func scope() {
 					let (signal, observer) = Signal<Never, Never>.pipe(disposable: disposable)

Tests/ReactiveSwiftTests/SignalProducerSpec.swift

@@ -849,9 +849,40 @@ class SignalProducerSpec: QuickSpec {
 
 					expect(result?.value) == [1, 4, 9, 16]
 				}
+
+				it("should interrupt all intermediate signals when the upstream is terminated") {
+					let baseProducer = SignalProducer<Int, Never>.empty
+
+					var lastEvent: Signal<Int, Never>.Event?
+
+					var isNeverEndingSignalDisposed = false
+
+					let disposable = baseProducer
+						.lift { _ in Signal.never.on(disposed: { isNeverEndingSignalDisposed = true }) }
+						.start { event in
+							lastEvent = event
+						}
+
+					expect(lastEvent).to(beNil())
+					expect(isNeverEndingSignalDisposed) == false
+
+					disposable.dispose()
+					expect(lastEvent) == .interrupted
+					expect(isNeverEndingSignalDisposed) == true
+				}
 			}
 
 			describe("over binary operators") {
+				var isNeverEndingSignalDisposed = false
+
+				let binaryLiftTransform = { (_: Signal<Int, Never>) -> (Signal<Int, Never>) -> Signal<Int, Never> in
+					{ (_: Signal<Int, Never>) in
+						Signal<Int, Never>.never.on(disposed: { isNeverEndingSignalDisposed = true })
+					}
+				}
+
+				beforeEach { isNeverEndingSignalDisposed = false }
+
 				it("should invoke transformation once per started signal") {
 					let baseProducer = SignalProducer<Int, Never>([1, 2])
 					let otherProducer = SignalProducer<Int, Never>([3, 4])
@@ -889,6 +920,44 @@ class SignalProducerSpec: QuickSpec {
 
 					expect(result?.value) == [5, 7, 9]
 				}
+
+				it("[left associative] should interrupt all intermediate signals when the upstream is terminated") {
+					let baseProducer = SignalProducer<Int, Never>.empty
+
+					var lastEvent: Signal<Int, Never>.Event?
+
+					let disposable = baseProducer
+						.liftLeft(binaryLiftTransform)(.empty)
+						.start { event in
+							lastEvent = event
+						}
+
+					expect(lastEvent).to(beNil())
+					expect(isNeverEndingSignalDisposed) == false
+
+					disposable.dispose()
+					expect(lastEvent) == .interrupted
+					expect(isNeverEndingSignalDisposed) == true
+				}
+
+				it("[right associative] should interrupt all intermediate signals when the upstream is terminated") {
+					let baseProducer = SignalProducer<Int, Never>.empty
+
+					var lastEvent: Signal<Int, Never>.Event?
+
+					let disposable = baseProducer
+						.liftRight(binaryLiftTransform)(.empty)
+						.start { event in
+							lastEvent = event
+						}
+
+					expect(lastEvent).to(beNil())
+					expect(isNeverEndingSignalDisposed) == false
+
+					disposable.dispose()
+					expect(lastEvent) == .interrupted
+					expect(isNeverEndingSignalDisposed) == true
+				}
 			}
 
 			describe("over binary operators with signal") {