Fixes array version of merge operator completing immediatelly in case one of the observable sequences is empty. ReactiveX#1221

kzaher · kzaher · commit 3e3af7eca6c4 · 2017-04-29T21:57:50.000+02:00
diff --git a/RxSwift/Observables/Merge.swift b/RxSwift/Observables/Merge.swift
@@ -442,12 +442,13 @@ fileprivate class MergeSink<SourceType, S: ObservableConvertibleType, O: Observe
 
     func run(_ sources: [SourceType]) -> Disposable {
         let _ = _group.insert(_sourceSubscription)
-        _stopped = true
 
         for source in sources {
             self.on(.next(source))
         }
 
+        _stopped = true
+
         checkCompleted()
 
         return _group
diff --git a/Sources/AllTestz/main.swift b/Sources/AllTestz/main.swift
@@ -1668,6 +1668,8 @@ final class ObservableMergeTest_ : ObservableMergeTest, RxTestCase {
     ("testMerge_MergeConcat_OuterError", ObservableMergeTest.testMerge_MergeConcat_OuterError),
     ("testMerge_MergeConcat_InnerError", ObservableMergeTest.testMerge_MergeConcat_InnerError),
     ("testMergeSync_Empty", ObservableMergeTest.testMergeSync_Empty),
+    ("testMergeSync_EmptyData_DoesntCompleteImmediatelly", ObservableMergeTest.testMergeSync_EmptyData_DoesntCompleteImmediatelly),
+    ("testMergeSync_EmptyEmpty_Completes", ObservableMergeTest.testMergeSync_EmptyEmpty_Completes),
     ("testMergeSync_Data", ObservableMergeTest.testMergeSync_Data),
     ("testMergeSync_ObservableOfObservable_InnerThrows", ObservableMergeTest.testMergeSync_ObservableOfObservable_InnerThrows),
     ("testFlatMapFirst_Complete", ObservableMergeTest.testFlatMapFirst_Complete),
diff --git a/Tests/RxSwiftTests/Observable+MergeTests.swift b/Tests/RxSwiftTests/Observable+MergeTests.swift
@@ -1015,6 +1015,68 @@ extension ObservableMergeTest {
             XCTAssertEqual(res.events, messages)
         }
     }
+
+    func testMergeSync_EmptyData_DoesntCompleteImmediatelly() {
+        let factories: [(Observable<Int>, Observable<Int>) -> Observable<Int>] =
+            [
+                { ys1, ys2 in Observable.merge(ys1, ys2) },
+                { ys1, ys2 in Observable.merge(AnyCollection([ys1, ys2])) },
+                { ys1, ys2 in Observable.merge([ys1, ys2]) },
+                ]
+        for factory in factories {
+            let scheduler = TestScheduler(initialClock: 0)
+
+            let ys1 = Observable<Int>.empty()
+
+            let ys2 = scheduler.createColdObservable([
+                next(10, 201),
+                next(20, 202),
+                completed(50)
+                ])
+
+            let res = scheduler.start {
+                factory(ys1.asObservable(), ys2.asObservable())
+            }
+
+            let messages = [
+                next(210, 201),
+                next(220, 202),
+                completed(250)
+            ]
+
+            XCTAssertEqual(res.events, messages)
+
+            XCTAssertEqual(ys2.subscriptions, [
+                Subscription(200, 250),
+                ])
+        }
+    }
+
+    func testMergeSync_EmptyEmpty_Completes() {
+        let factories: [(Observable<Int>, Observable<Int>) -> Observable<Int>] =
+            [
+                { ys1, ys2 in Observable.merge(ys1, ys2) },
+                { ys1, ys2 in Observable.merge(AnyCollection([ys1, ys2])) },
+                { ys1, ys2 in Observable.merge([ys1, ys2]) },
+                ]
+        for factory in factories {
+            let scheduler = TestScheduler(initialClock: 0)
+
+            let ys1 = Observable<Int>.empty()
+
+            let ys2 = Observable<Int>.empty()
+
+            let res = scheduler.start {
+                factory(ys1.asObservable(), ys2.asObservable())
+            }
+
+            let messages = [
+                completed(200, Int.self)
+            ]
+
+            XCTAssertEqual(res.events, messages)
+        }
+    }
     
     func testMergeSync_Data() {
         let factories: [(Observable<Int>, Observable<Int>, Observable<Int>) -> Observable<Int>] =
