Borrow the start and observe overloads from neilpa/Rex

Author: ikesyo

ReactiveTask.xcodeproj/project.pbxproj

@@ -7,6 +7,7 @@
 	objects = {
 
 /* Begin PBXBuildFile section */
+		CD0E570A1BBD2A31009BB2AE /* Extensions.swift in Sources */ = {isa = PBXBuildFile; fileRef = CD0E57091BBD2A31009BB2AE /* Extensions.swift */; settings = {ASSET_TAGS = (); }; };
 		D02130921AF87B6500B9EC20 /* Result.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = D02130911AF87B6500B9EC20 /* Result.framework */; };
 		D0BFEA5E1A2D1E5E00E23194 /* ReactiveTask.h in Headers */ = {isa = PBXBuildFile; fileRef = D0BFEA5D1A2D1E5E00E23194 /* ReactiveTask.h */; settings = {ATTRIBUTES = (Public, ); }; };
 		D0BFEA641A2D1E5E00E23194 /* ReactiveTask.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = D0BFEA581A2D1E5E00E23194 /* ReactiveTask.framework */; };
@@ -30,6 +31,7 @@
 /* End PBXContainerItemProxy section */
 
 /* Begin PBXFileReference section */
+		CD0E57091BBD2A31009BB2AE /* Extensions.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = Extensions.swift; sourceTree = "<group>"; };
 		D02130911AF87B6500B9EC20 /* Result.framework */ = {isa = PBXFileReference; lastKnownFileType = wrapper.framework; path = Result.framework; sourceTree = BUILT_PRODUCTS_DIR; };
 		D0BFEA581A2D1E5E00E23194 /* ReactiveTask.framework */ = {isa = PBXFileReference; explicitFileType = wrapper.framework; includeInIndex = 0; path = ReactiveTask.framework; sourceTree = BUILT_PRODUCTS_DIR; };
 		D0BFEA5C1A2D1E5E00E23194 /* Info.plist */ = {isa = PBXFileReference; lastKnownFileType = text.plist.xml; path = Info.plist; sourceTree = "<group>"; };
@@ -111,6 +113,7 @@
 			children = (
 				D0BFEA5D1A2D1E5E00E23194 /* ReactiveTask.h */,
 				D0BFEAA31A2D216600E23194 /* Errors.swift */,
+				CD0E57091BBD2A31009BB2AE /* Extensions.swift */,
 				D0BFEA9F1A2D212000E23194 /* Task.swift */,
 				D0BFEA5B1A2D1E5E00E23194 /* Supporting Files */,
 			);
@@ -321,6 +324,7 @@
 			buildActionMask = 2147483647;
 			files = (
 				D0BFEAA41A2D216600E23194 /* Errors.swift in Sources */,
+				CD0E570A1BBD2A31009BB2AE /* Extensions.swift in Sources */,
 				D0BFEAA01A2D212000E23194 /* Task.swift in Sources */,
 			);
 			runOnlyForDeploymentPostprocessing = 0;

ReactiveTask/Extensions.swift

@@ -0,0 +1,51 @@
+//
+//  Extensions.swift
+//  ReactiveTask
+//
+//  Created by Syo Ikeda on 10/1/15.
+//  Copyright © 2015 Carthage. All rights reserved.
+//
+
+import ReactiveCocoa
+
+extension Signal {
+
+	/// Bring back the `observe` overload. The `observeNext` or pattern matching
+	/// on `observe(Event)` is still annoying in practice and more verbose. This is
+	/// also likely to change in a later RAC 4 alpha.
+	internal func observe(next next: (T -> ())? = nil, error: (E -> ())? = nil, completed: (() -> ())? = nil, interrupted: (() -> ())? = nil) -> Disposable? {
+		return self.observe { (event: Event<T, E>) in
+			switch event {
+			case let .Next(value):
+				next?(value)
+			case let .Error(err):
+				error?(err)
+			case .Completed:
+				completed?()
+			case .Interrupted:
+				interrupted?()
+			}
+		}
+	}
+}
+
+extension SignalProducer {
+
+	/// Bring back the `start` overload. The `startNext` or pattern matching
+	/// on `start(Event)` is annoying in practice and more verbose. This is also
+	/// likely to change in a later RAC 4 alpha.
+	internal func start(next next: (T -> ())? = nil, error: (E -> ())? = nil, completed: (() -> ())? = nil, interrupted: (() -> ())? = nil) -> Disposable? {
+		return self.start { (event: Event<T, E>) in
+			switch event {
+			case let .Next(value):
+				next?(value)
+			case let .Error(err):
+				error?(err)
+			case .Completed:
+				completed?()
+			case .Interrupted:
+				interrupted?()
+			}
+		}
+	}
+}

ReactiveTask/Task.swift

@@ -184,26 +184,21 @@ private final class Pipe {
 			producer.startWithSignal { signal, producerDisposable in
 				disposable.addDisposable(producerDisposable)
 
-				signal.observe {
-					switch $0 {
-					case let .Next(data):
-						let dispatchData = dispatch_data_create(data.bytes, data.length, self.queue, nil)
-						
-						dispatch_io_write(channel, 0, dispatchData, self.queue) { (done, data, error) in
-							if error == ECANCELED {
-								sendInterrupted(observer)
-							} else if error != 0 {
-								sendError(observer, .POSIXError(error))
-							}
+				signal.observe(next: { data in
+					let dispatchData = dispatch_data_create(data.bytes, data.length, self.queue, nil)
+
+					dispatch_io_write(channel, 0, dispatchData, self.queue) { (done, data, error) in
+						if error == ECANCELED {
+							sendInterrupted(observer)
+						} else if error != 0 {
+							sendError(observer, .POSIXError(error))
 						}
-					case .Completed:
-						dispatch_io_close(channel, 0)
-					case .Interrupted:
-						sendInterrupted(observer)
-					default:
-						break
 					}
-				}
+				}, completed: {
+					dispatch_io_close(channel, 0)
+				}, interrupted: {
+					sendInterrupted(observer)
+				})
 			}
 
 			disposable.addDisposable {
@@ -249,25 +244,22 @@ private func aggregateDataReadFromPipe(pipe: Pipe) -> SignalProducer<ReadData, R
 		readProducer.startWithSignal { signal, signalDisposable in
 			disposable.addDisposable(signalDisposable)
 
-			signal.observe {
-				switch $0 {
-				case let .Next(data):
-					sendNext(observer, .chunk(data))
+			signal.observe(next: { data in
+				sendNext(observer, .chunk(data))
 
-					if let existingBuffer = buffer {
-						buffer = dispatch_data_create_concat(existingBuffer, data)
-					} else {
-						buffer = data
-					}
-				case let .Error(error):
-					sendError(observer, error)
-				case .Completed:
-					sendNext(observer, .aggregated(buffer))
-					sendCompleted(observer)
-				case .Interrupted:
-					sendInterrupted(observer)
+				if let existingBuffer = buffer {
+					buffer = dispatch_data_create_concat(existingBuffer, data)
+				} else {
+					buffer = data
 				}
-			}
+			}, error: { error in
+				sendError(observer, error)
+			}, completed: {
+				sendNext(observer, .aggregated(buffer))
+				sendCompleted(observer)
+			}, interrupted: {
+				sendInterrupted(observer)
+			})
 		}
 	}
 }
@@ -454,49 +446,43 @@ public func launchTask(taskDescription: TaskDescription) -> SignalProducer<TaskE
 					stdoutProducer.startWithSignal { signal, signalDisposable in
 						disposable += signalDisposable
 
-						signal.observe {
-							switch $0 {
-							case let .Next(readData):
-								switch readData {
-								case let .Chunk(data):
-									sendNext(observer, .StandardOutput(data))
+						signal.observe(next: { readData in
+							switch readData {
+							case let .Chunk(data):
+								sendNext(observer, .StandardOutput(data))
 
-								case let .Aggregated(data):
-									sendNext(stdoutAggregatedSink, data)
-								}
-							case let .Error(error):
-								sendError(observer, error)
-								sendError(stdoutAggregatedSink, error)
-							case .Completed:
-								sendCompleted(stdoutAggregatedSink)
-							case .Interrupted:
-								sendInterrupted(stdoutAggregatedSink)
+							case let .Aggregated(data):
+								sendNext(stdoutAggregatedSink, data)
 							}
-						}
+						}, error: { error in
+							sendError(observer, error)
+							sendError(stdoutAggregatedSink, error)
+						}, completed: {
+							sendCompleted(stdoutAggregatedSink)
+						}, interrupted: {
+							sendInterrupted(stdoutAggregatedSink)
+						})
 					}
 
 					stderrProducer.startWithSignal { signal, signalDisposable in
 						disposable += signalDisposable
 
-						signal.observe {
-							switch $0 {
-							case let .Next(readData):
-								switch readData {
-								case let .Chunk(data):
-									sendNext(observer, .StandardError(data))
+						signal.observe(next: { readData in
+							switch readData {
+							case let .Chunk(data):
+								sendNext(observer, .StandardError(data))
 
-								case let .Aggregated(data):
-									sendNext(stderrAggregatedSink, data)
-								}
-							case let .Error(error):
-								sendError(observer, error)
-								sendError(stderrAggregatedSink, error)
-							case .Completed:
-								sendCompleted(stderrAggregatedSink)
-							case .Interrupted:
-								sendInterrupted(stderrAggregatedSink)
+							case let .Aggregated(data):
+								sendNext(stderrAggregatedSink, data)
 							}
-						}
+						}, error: { error in
+							sendError(observer, error)
+							sendError(stderrAggregatedSink, error)
+						}, completed: {
+							sendCompleted(stderrAggregatedSink)
+						}, interrupted: {
+							sendInterrupted(stderrAggregatedSink)
+						})
 					}
 
 					task.standardOutput = stdoutPipe.writeHandle