Signal resource management with Lifetime. (#404)

* Use Lifetime to manage Signal disposables.

* Clean up a few instances of disposable lifetime association.

* Update the changelog for #404. [skip ci]

Author: andersio

CHANGELOG.md

@@ -1,7 +1,23 @@
 # master
 *Please add new entries at the top.*
 
-1. `SignalProducer.startWithSignal` now returns the value of the setup closure.
+1. `Signal` now uses `Lifetime` for resource management. (#404, kudos to @andersio)
+
+   The `Signal` initialzer now accepts a generator closure that is passed with the input `Observer` and the `Lifetime` as its arguments. The original variant accepting a single-argument generator closure is now obselete. This is a source breaking change.
+   
+   ```swift
+   // New: Add `Disposable`s to the `Lifetime`.
+   let candies = Signal<U, E> { (observer: Signal<U, E>.Observer, lifetime: Lifetime) in
+      lifetime += trickOrTreat.observe(observer)
+   }
+   
+   // Obsolete: Returning a `Disposable`.
+   let candies = Signal { (observer: Signal<U, E>.Observer) -> Disposable? in
+      return trickOrTreat.observe(observer)
+   }
+   ```
+
+1. `SignalProducer.startWithSignal` now returns the value of the setup closure. (#533, kudos to @Burgestrand)
 
 # 2.1.0-alpha.2
 1. Disabled code coverage data to allow app submissions with Xcode 9.0 (see https://github.com/Carthage/Carthage/issues/2056, kudos to @NachoSoto)

Sources/Deprecations+Removals.swift

@@ -11,6 +11,9 @@ extension AnyDisposable {
 extension Signal {
 	@available(*, unavailable, renamed:"promoteError")
 	public func promoteErrors<F>(_: F.Type) -> Signal<Value, F> { fatalError() }
+
+	@available(*, unavailable, message:"Use the `Signal.init` that accepts a two-argument generator.")
+	public convenience init(_ generator: (Observer) -> Disposable?) { fatalError() }
 }
 
 extension SignalProducer {

Sources/Flatten.swift

@@ -288,18 +288,12 @@ extension Signal where Value: SignalProducerConvertible, Error == Value.Error {
 	fileprivate func concurrent(limit: UInt) -> Signal<Value.Value, Error> {
 		precondition(limit > 0, "The concurrent limit must be greater than zero.")
 
-		return Signal<Value.Value, Error> { relayObserver in
-			let disposable = CompositeDisposable()
-			let relayDisposable = CompositeDisposable()
-
-			disposable += relayDisposable
-			disposable += self.observeConcurrent(relayObserver, limit, relayDisposable)
-
-			return disposable
+		return Signal<Value.Value, Error> { relayObserver, lifetime in
+			lifetime += self.observeConcurrent(relayObserver, limit, lifetime)
 		}
 	}
 
-	fileprivate func observeConcurrent(_ observer: Signal<Value.Value, Error>.Observer, _ limit: UInt, _ disposable: CompositeDisposable) -> Disposable? {
+	fileprivate func observeConcurrent(_ observer: Signal<Value.Value, Error>.Observer, _ limit: UInt, _ lifetime: Lifetime) -> Disposable? {
 		let state = Atomic(ConcurrentFlattenState<Value.Value, Error>(limit: limit))
 
 		func startNextIfNeeded() {
@@ -308,7 +302,7 @@ extension Signal where Value: SignalProducerConvertible, Error == Value.Error {
 				let deinitializer = ScopedDisposable(AnyDisposable(producerState.deinitialize))
 
 				producer.startWithSignal { signal, inner in
-					let handle = disposable.add(inner)
+					let handle = lifetime.observeEnded(inner.dispose)
 
 					signal.observe { event in
 						switch event {
@@ -371,12 +365,10 @@ extension SignalProducer where Value: SignalProducerConvertible, Error == Value.
 		precondition(limit > 0, "The concurrent limit must be greater than zero.")
 
 		return SignalProducer<Value.Value, Error> { relayObserver, lifetime in
-			self.startWithSignal { signal, signalDisposable in
-				let disposables = CompositeDisposable()
-				lifetime.observeEnded(signalDisposable.dispose)
-				lifetime.observeEnded(disposables.dispose)
+			self.startWithSignal { signal, interruptHandle in
+				lifetime.observeEnded(interruptHandle.dispose)
 
-				_ = signal.observeConcurrent(relayObserver, limit, disposables)
+				_ = signal.observeConcurrent(relayObserver, limit, lifetime)
 			}
 		}
 	}
@@ -533,14 +525,10 @@ extension Signal where Value: SignalProducerConvertible, Error == Value.Error {
 	/// - note: The returned signal completes when `signal` and the latest inner
 	///         signal have both completed.
 	fileprivate func switchToLatest() -> Signal<Value.Value, Error> {
-		return Signal<Value.Value, Error> { observer in
-			let composite = CompositeDisposable()
+		return Signal<Value.Value, Error> { observer, lifetime in
 			let serial = SerialDisposable()
-
-			composite += serial
-			composite += self.observeSwitchToLatest(observer, serial)
-
-			return composite
+			lifetime += serial
+			lifetime += self.observeSwitchToLatest(observer, serial)
 		}
 	}
 
@@ -658,14 +646,10 @@ extension Signal where Value: SignalProducerConvertible, Error == Value.Error {
 	///
 	/// The returned signal completes when `self` and the winning inner signal have both completed.
 	fileprivate func race() -> Signal<Value.Value, Error> {
-		return Signal<Value.Value, Error> { observer in
-			let composite = CompositeDisposable()
+		return Signal<Value.Value, Error> { observer, lifetime in
 			let relayDisposable = CompositeDisposable()
-
-			composite += relayDisposable
-			composite += self.observeRace(observer, relayDisposable)
-
-			return composite
+			lifetime += relayDisposable
+			lifetime += self.observeRace(observer, relayDisposable)
 		}
 	}
 
@@ -905,8 +889,8 @@ extension Signal {
 	///   - transform: A closure that accepts emitted error and returns a signal
 	///                producer with a different type of error.
 	public func flatMapError<F>(_ transform: @escaping (Error) -> SignalProducer<Value, F>) -> Signal<Value, F> {
-		return Signal<Value, F> { observer in
-			self.observeFlatMapError(transform, observer, SerialDisposable())
+		return Signal<Value, F> { observer, lifetime in
+			lifetime += self.observeFlatMapError(transform, observer, SerialDisposable())
 		}
 	}
 

Sources/FoundationExtensions.swift

@@ -30,12 +30,12 @@ extension Reactive where Base: NotificationCenter {
 	/// - note: The signal does not terminate naturally. Observers must be
 	///         explicitly disposed to avoid leaks.
 	public func notifications(forName name: Notification.Name?, object: AnyObject? = nil) -> Signal<Notification, NoError> {
-		return Signal { [base = self.base] observer in
+		return Signal { [base = self.base] observer, lifetime in
 			let notificationObserver = base.addObserver(forName: name, object: object, queue: nil) { notification in
 				observer.send(value: notification)
 			}
 
-			return AnyDisposable {
+			lifetime.observeEnded {
 				base.removeObserver(notificationObserver)
 			}
 		}

Sources/Lifetime.swift

@@ -11,8 +11,8 @@ public final class Lifetime {
 	/// - note: Consider using `Lifetime.observeEnded` if only a closure observer
 	///         is to be attached.
 	public var ended: Signal<Never, NoError> {
-		return Signal { observer in
-			return disposables += observer.sendCompleted
+		return Signal { observer, lifetime in
+			lifetime += (disposables += observer.sendCompleted)
 		}
 	}
 
@@ -63,7 +63,8 @@ public final class Lifetime {
 	///            if `lifetime` has already ended.
 	@discardableResult
 	public static func += (lifetime: Lifetime, disposable: Disposable?) -> Disposable? {
-		return (disposable?.dispose).flatMap(lifetime.observeEnded)
+		guard let dispose = disposable?.dispose else { return nil }
+		return lifetime.observeEnded(dispose)
 	}
 }