Update ReactiveSwift to 2.0.0-rc.1

Cartfile

@@ -1,2 +1,2 @@
 github "antitypical/Result" ~> 3.2.1
-github "ReactiveCocoa/ReactiveSwift" ~> 1.1
+github "ReactiveCocoa/ReactiveSwift" "2.0.0-rc.1"

Cartfile.resolved

@@ -1,5 +1,5 @@
 github "Quick/Nimble" "v7.0.1"
 github "Quick/Quick" "v1.1.0"
-github "ReactiveCocoa/ReactiveSwift" "1.1.3"
+github "ReactiveCocoa/ReactiveSwift" "2.0.0-rc.1"
 github "antitypical/Result" "3.2.3"
 github "jspahrsummers/xcconfigs" "3d9d99634cae6d586e272543d527681283b33eb0"

Package.swift

@@ -10,7 +10,7 @@ let package = Package(
     dependencies: {
         var deps: [Package.Dependency] = [
             .Package(url: "https://github.com/antitypical/Result.git", versions: Version(3, 2, 1)..<Version(3, .max, .max)),
-            .Package(url: "https://github.com/ReactiveCocoa/ReactiveSwift.git", versions: Version(1, 1, 0)..<Version(1, .max, .max)),
+            .Package(url: "https://github.com/ReactiveCocoa/ReactiveSwift.git", "2.0.0-rc.1"),
         ]
         if isSwiftPackagerManagerTest {
             deps += [

Sources/Task.swift

@@ -162,7 +162,7 @@ private final class Pipe {
 	/// After starting the returned producer, `readFD` should not be used
 	/// anywhere else, as it may close unexpectedly.
 	func transferReadsToProducer() -> ReadProducer {
-		return SignalProducer { observer, disposable in
+		return SignalProducer { observer, lifetime in
 			self.group.enter()
 			let channel = DispatchIO(type: .stream, fileDescriptor: self.readFD, queue: self.queue) { error in
 				if error == 0 {
@@ -200,7 +200,7 @@ private final class Pipe {
 				}
 			}
 
-			let _ = disposable.add {
+			lifetime.observeEnded {
 				channel.close(flags: .stop)
 			}
 		}
@@ -215,7 +215,7 @@ private final class Pipe {
 	///
 	/// Returns a producer that will complete or error.
 	func writeDataFromProducer(_ producer: SignalProducer<Data, NoError>) -> SignalProducer<(), TaskError> {
-		return SignalProducer { observer, disposable in
+		return SignalProducer { observer, lifetime in
 			self.group.enter()
 			let channel = DispatchIO(type: .stream, fileDescriptor: self.writeFD, queue: self.queue) { error in
 				if error == 0 {
@@ -231,9 +231,9 @@ private final class Pipe {
 			}
 
 			producer.startWithSignal { signal, producerDisposable in
-				disposable.add(producerDisposable)
+				lifetime.observeEnded(producerDisposable.dispose)
 
-				signal.observe(Observer(value: { data in
+				signal.observe(Signal.Observer(value: { data in
 					let dispatchData = data.withUnsafeBytes { (bytes: UnsafePointer<UInt8>) -> DispatchData in
 						let buffer = UnsafeBufferPointer(start: bytes, count: data.count)
 						return DispatchData(bytes: buffer)
@@ -253,7 +253,7 @@ private final class Pipe {
 				}))
 			}
 
-			let _ = disposable.add {
+			lifetime.observeEnded {
 				channel.close(flags: .stop)
 			}
 		}
@@ -409,7 +409,7 @@ extension Task {
 	/// - Returns: A producer that will launch the receiver when started, then send
 	///            `TaskEvent`s as execution proceeds.
 	public func launch(standardInput: SignalProducer<Data, NoError>? = nil) -> SignalProducer<TaskEvent<Data>, TaskError> {
-		return SignalProducer { observer, disposable in
+		return SignalProducer { observer, lifetime in
 			let queue = DispatchQueue(label: self.description, attributes: [])
 			let group = Task.group
 
@@ -466,15 +466,15 @@ extension Task {
 						}
 					}
 
-					return SignalProducer { observer, disposable in
+					return SignalProducer { observer, lifetime in
 						func startAggregating(producer: Pipe.ReadProducer, chunk: @escaping (Data) -> TaskEvent<Data>) -> Pipe.ReadProducer {
 							let aggregated = MutableProperty<Aggregation?>(nil)
 
 							producer.startWithSignal { signal, signalDisposable in
-								disposable += signalDisposable
+								lifetime.observeEnded(signalDisposable.dispose)
 
 								var aggregate = Data()
-								signal.observe(Observer(value: { data in
+								signal.observe(Signal.Observer(value: { data in
 									observer.send(value: chunk(data))
 									aggregate.append(data)
 								}, failed: { error in
@@ -504,20 +504,22 @@ extension Task {
 							if terminationStatus == EXIT_SUCCESS {
 								// Wait for stderr to finish, then pass
 								// through stdout.
-								disposable += stderrAggregated
+								let disposable = stderrAggregated
 									.then(stdoutAggregated)
 									.map(TaskEvent.success)
 									.start(observer)
+								lifetime.observeEnded(disposable.dispose)
 							} else {
 								// Wait for stdout to finish, then pass
 								// through stderr.
-								disposable += stdoutAggregated
+								let disposable = stdoutAggregated
 									.then(stderrAggregated)
 									.flatMap(.concat) { data -> SignalProducer<TaskEvent<Data>, TaskError> in
 										let errorString = (data.count > 0 ? String(data: data, encoding: .utf8) : nil)
 										return SignalProducer(error: .shellTaskFailed(self, exitCode: terminationStatus, standardError: errorString))
 									}
 									.start(observer)
+								lifetime.observeEnded(disposable.dispose)
 							}
 							group.leave()
 						}
@@ -527,15 +529,15 @@ extension Task {
 						close(stdoutPipe.writeFD)
 						close(stderrPipe.writeFD)
 
-						disposable += stdinProducer.start()
+						lifetime.observeEnded(stdinProducer.start().dispose)
 
-						let _ = disposable.add {
+						lifetime.observeEnded {
 							process.terminate()
 						}
 					}
 				}
 				.startWithSignal { signal, taskDisposable in
-					disposable.add(taskDisposable)
+					lifetime.observeEnded(taskDisposable.dispose)
 					signal.observe(observer)
 				}
 		}