Add merge(with:) operator

CHANGELOG.md

@@ -1,6 +1,7 @@
 # master
 *Please add new entries at the top.*
 
+1. New operator `merge(with:)` (#600, kudos to @ra1028)
 1. New operator `map(value:)` (#601, kudos to @ra1028)
 
 # 3.1.0

Sources/Flatten.swift

@@ -492,19 +492,19 @@ extension Signal {
 			.flatten(.merge)
 			.startAndRetrieveSignal()
 	}
-
+	
 	/// Merges the given signals into a single `Signal` that will emit all
 	/// values from each of them, and complete when all of them have completed.
 	///
 	/// - parameters:
-    ///   - signals: A list of signals to merge.
+	///   - signals: A list of signals to merge.
 	public static func merge(_ signals: Signal<Value, Error>...) -> Signal<Value, Error> {
 		return Signal.merge(signals)
 	}
 }
 
 extension SignalProducer {
-	/// Merges the given producers into a single `SignalProducer` that will emit
+ 	/// Merges the given producers into a single `SignalProducer` that will emit
 	/// all values from each of them, and complete when all of them have
 	/// completed.
 	///
@@ -514,7 +514,7 @@ extension SignalProducer {
 	{
 		return SignalProducer<Seq.Iterator.Element, NoError>(producers).flatten(.merge)
 	}
-
+	
 	/// Merges the given producers into a single `SignalProducer` that will emit
 	/// all values from each of them, and complete when all of them have
 	/// completed.

Sources/Signal.swift

@@ -798,6 +798,17 @@ extension Signal {
 	public func combineLatest<U>(with other: Signal<U, Error>) -> Signal<(Value, U), Error> {
 		return Signal.combineLatest(self, other)
 	}
+
+	/// Merge the given signal into a single `Signal` that will emit all
+	/// values from both of them, and complete when all of them have completed.
+	///
+	/// - parameters:
+	///   - other: A signal to merge `self`'s value with.
+	///
+	/// - returns: A signal that sends all values of `self` and given signal.
+	public func merge(with other: Signal<Value, Error>) -> Signal<Value, Error> {
+		return Signal.merge(self, other)
+	}
 
 	/// Delay `value` and `completed` events by the given interval, forwarding
 	/// them on the given scheduler.

Sources/SignalProducer.swift

@@ -1087,6 +1087,17 @@ extension SignalProducer {
 	public func combineLatest<Other: SignalProducerConvertible>(with other: Other) -> SignalProducer<(Value, Other.Value), Error> where Other.Error == Error {
 		return SignalProducer.combineLatest(self, other)
 	}
+
+	/// Merge the given producer into a single `SignalProducer` that will emit all
+	/// values from both of them, and complete when all of them have completed.
+	///
+	/// - parameters:
+	///   - other: A producer to merge `self`'s value with.
+	///
+	/// - returns: A producer that sends all values of `self` and given producer.
+	public func merge(with other: SignalProducer<Value, Error>) -> SignalProducer<Value, Error> {
+		return SignalProducer.merge(self, other)
+	}
 
 	/// Delay `value` and `completed` events by the given interval, forwarding
 	/// them on the given scheduler.

Tests/ReactiveSwiftTests/FlattenSpec.swift

@@ -820,8 +820,9 @@ class FlattenSpec: QuickSpec {
 			it("should emit values from all signals") {
 				let (signal1, observer1) = Signal<Int, NoError>.pipe()
 				let (signal2, observer2) = Signal<Int, NoError>.pipe()
+				let (signal3, observer3) = Signal<Int, NoError>.pipe()
 
-				let mergedSignals = Signal.merge([signal1, signal2])
+				let mergedSignals = Signal.merge([signal1, signal2, signal3])
 
 				var lastValue: Int?
 				mergedSignals.observeValues { lastValue = $0 }
@@ -834,15 +835,19 @@ class FlattenSpec: QuickSpec {
 				observer2.send(value: 2)
 				expect(lastValue) == 2
 
-				observer1.send(value: 3)
+				observer3.send(value: 3)
 				expect(lastValue) == 3
+
+				observer1.send(value: 4)
+				expect(lastValue) == 4
 			}
 
 			it("should not stop when one signal completes") {
 				let (signal1, observer1) = Signal<Int, NoError>.pipe()
 				let (signal2, observer2) = Signal<Int, NoError>.pipe()
+				let (signal3, observer3) = Signal<Int, NoError>.pipe()
 
-				let mergedSignals = Signal.merge([signal1, signal2])
+				let mergedSignals = Signal.merge([signal1, signal2, signal3])
 
 				var lastValue: Int?
 				mergedSignals.observeValues { lastValue = $0 }
@@ -857,13 +862,17 @@ class FlattenSpec: QuickSpec {
 
 				observer2.send(value: 2)
 				expect(lastValue) == 2
+
+				observer3.send(value: 3)
+				expect(lastValue) == 3
 			}
 
 			it("should complete when all signals complete") {
 				let (signal1, observer1) = Signal<Int, NoError>.pipe()
 				let (signal2, observer2) = Signal<Int, NoError>.pipe()
+				let (signal3, observer3) = Signal<Int, NoError>.pipe()
 
-				let mergedSignals = Signal.merge([signal1, signal2])
+				let mergedSignals = Signal.merge([signal1, signal2, signal3])
 
 				var completed = false
 				mergedSignals.observeCompleted { completed = true }
@@ -876,71 +885,216 @@ class FlattenSpec: QuickSpec {
 				observer1.sendCompleted()
 				expect(completed) == false
 
+				observer2.sendCompleted()
+				expect(completed) == false
+
+				observer3.sendCompleted()
+				expect(completed) == true
+			}
+		}
+
+		describe("Signal.merge(with:)") {
+			it("should emit values from both signals") {
+				let (signal1, observer1) = Signal<Int, NoError>.pipe()
+				let (signal2, observer2) = Signal<Int, NoError>.pipe()
+
+				let mergedSignals = signal1.merge(with: signal2)
+
+				var lastValue: Int?
+				mergedSignals.observeValues { lastValue = $0 }
+
+				expect(lastValue).to(beNil())
+
+				observer1.send(value: 1)
+				expect(lastValue) == 1
+
+				observer2.send(value: 2)
+				expect(lastValue) == 2
+
+				observer1.send(value: 3)
+				expect(lastValue) == 3
+			}
+
+			it("should not stop when one signal completes") {
+				let (signal1, observer1) = Signal<Int, NoError>.pipe()
+				let (signal2, observer2) = Signal<Int, NoError>.pipe()
+
+				let mergedSignals = signal1.merge(with: signal2)
+
+				var lastValue: Int?
+				mergedSignals.observeValues { lastValue = $0 }
+
+				expect(lastValue).to(beNil())
+
+				observer1.send(value: 1)
+				expect(lastValue) == 1
+
+				observer1.sendCompleted()
+				expect(lastValue) == 1
+
+				observer2.send(value: 2)
+				expect(lastValue) == 2
+			}
+
+			it("should complete when both signals complete") {
+				let (signal1, observer1) = Signal<Int, NoError>.pipe()
+				let (signal2, observer2) = Signal<Int, NoError>.pipe()
+
+				let mergedSignals = signal1.merge(with: signal2)
+
+				var completed = false
+				mergedSignals.observeCompleted { completed = true }
+
+				expect(completed) == false
+
+				observer1.send(value: 1)
+				expect(completed) == false
+
+				observer1.sendCompleted()
+				expect(completed) == false
+
 				observer2.sendCompleted()
 				expect(completed) == true
 			}
 		}
 
 		describe("SignalProducer.merge()") {
 			it("should emit values from all producers") {
-				let (signal1, observer1) = SignalProducer<Int, NoError>.pipe()
-				let (signal2, observer2) = SignalProducer<Int, NoError>.pipe()
+				let (producer1, observer1) = SignalProducer<Int, NoError>.pipe()
+				let (producer2, observer2) = SignalProducer<Int, NoError>.pipe()
+				let (producer3, observer3) = SignalProducer<Int, NoError>.pipe()
 
-				let mergedSignals = SignalProducer.merge([signal1, signal2])
+				let mergedProducer = SignalProducer.merge([producer1, producer2, producer3])
 
 				var lastValue: Int?
-				mergedSignals.startWithValues { lastValue = $0 }
+				mergedProducer.startWithValues { lastValue = $0 }
 
 				expect(lastValue).to(beNil())
 
 				observer1.send(value: 1)
 				expect(lastValue) == 1
-
+				
 				observer2.send(value: 2)
 				expect(lastValue) == 2
-
-				observer1.send(value: 3)
+				
+				observer3.send(value: 3)
 				expect(lastValue) == 3
+
+				observer1.send(value: 4)
+				expect(lastValue) == 4
 			}
 
 			it("should not stop when one producer completes") {
-				let (signal1, observer1) = SignalProducer<Int, NoError>.pipe()
-				let (signal2, observer2) = SignalProducer<Int, NoError>.pipe()
+				let (producer1, observer1) = SignalProducer<Int, NoError>.pipe()
+				let (producer2, observer2) = SignalProducer<Int, NoError>.pipe()
+				let (producer3, observer3) = SignalProducer<Int, NoError>.pipe()
 
-				let mergedSignals = SignalProducer.merge([signal1, signal2])
+				let mergedProducer = SignalProducer.merge([producer1, producer2, producer3])
 
 				var lastValue: Int?
-				mergedSignals.startWithValues { lastValue = $0 }
+				mergedProducer.startWithValues { lastValue = $0 }
 
 				expect(lastValue).to(beNil())
 
 				observer1.send(value: 1)
 				expect(lastValue) == 1
-
+				
 				observer1.sendCompleted()
 				expect(lastValue) == 1
-
+				
 				observer2.send(value: 2)
 				expect(lastValue) == 2
+
+				observer3.send(value: 3)
+				expect(lastValue) == 3
 			}
 
 			it("should complete when all producers complete") {
-				let (signal1, observer1) = SignalProducer<Int, NoError>.pipe()
-				let (signal2, observer2) = SignalProducer<Int, NoError>.pipe()
+				let (producer1, observer1) = SignalProducer<Int, NoError>.pipe()
+				let (producer2, observer2) = SignalProducer<Int, NoError>.pipe()
+				let (producer3, observer3) = SignalProducer<Int, NoError>.pipe()
 
-				let mergedSignals = SignalProducer.merge([signal1, signal2])
+				let mergedProducer = SignalProducer.merge([producer1, producer2, producer3])
 
 				var completed = false
-				mergedSignals.startWithCompleted { completed = true }
+				mergedProducer.startWithCompleted { completed = true }
 
 				expect(completed) == false
-
+				
 				observer1.send(value: 1)
 				expect(completed) == false
-
+				
 				observer1.sendCompleted()
 				expect(completed) == false
-
+
+				observer2.sendCompleted()
+				expect(completed) == false
+
+				observer3.sendCompleted()
+				expect(completed) == true
+			}
+		}
+
+		describe("SignalProducer.merge(with:)") {
+			it("should emit values from both producers") {
+				let (producer1, observer1) = SignalProducer<Int, NoError>.pipe()
+				let (producer2, observer2) = SignalProducer<Int, NoError>.pipe()
+
+				let mergedProducer = producer1.merge(with: producer2)
+
+				var lastValue: Int?
+				mergedProducer.startWithValues { lastValue = $0 }
+
+				expect(lastValue).to(beNil())
+
+				observer1.send(value: 1)
+				expect(lastValue) == 1
+
+				observer2.send(value: 2)
+				expect(lastValue) == 2
+
+				observer1.send(value: 3)
+				expect(lastValue) == 3
+			}
+
+			it("should not stop when one producer completes") {
+				let (producer1, observer1) = SignalProducer<Int, NoError>.pipe()
+				let (producer2, observer2) = SignalProducer<Int, NoError>.pipe()
+
+				let mergedProducer = producer1.merge(with: producer2)
+
+				var lastValue: Int?
+				mergedProducer.startWithValues { lastValue = $0 }
+
+				expect(lastValue).to(beNil())
+
+				observer1.send(value: 1)
+				expect(lastValue) == 1
+
+				observer1.sendCompleted()
+				expect(lastValue) == 1
+
+				observer2.send(value: 2)
+				expect(lastValue) == 2
+			}
+
+			it("should complete when both producers complete") {
+				let (producer1, observer1) = SignalProducer<Int, NoError>.pipe()
+				let (producer2, observer2) = SignalProducer<Int, NoError>.pipe()
+
+				let mergedProducer = producer1.merge(with: producer2)
+
+				var completed = false
+				mergedProducer.startWithCompleted { completed = true }
+
+				expect(completed) == false
+
+				observer1.send(value: 1)
+				expect(completed) == false
+
+				observer1.sendCompleted()
+				expect(completed) == false
+
 				observer2.sendCompleted()
 				expect(completed) == true
 			}