Improves locking behavior of Observable.merge (optimized array versio…

…n). #1344
ReactiveX · Jul 19, 2017 · 37a44d9 · 37a44d9
1 parent 79ac3d0
commit 37a44d9
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 4 deletions.
diff --git a/RxSwift/Observables/Merge.swift b/RxSwift/Observables/Merge.swift
@@ -462,6 +462,7 @@ fileprivate class MergeSink<SourceElement, SourceSequence: ObservableConvertible
                 }
                 do {
                     let value = try performMap(element)
+                    _activeCount += 1
                     subscribeInner(value.asObservable())
                 }
                 catch let e {
@@ -482,18 +483,17 @@ fileprivate class MergeSink<SourceElement, SourceSequence: ObservableConvertible
     func subscribeInner(_ source: Observable<Observer.E>) {
         let iterDisposable = SingleAssignmentDisposable()
         if let disposeKey = _group.insert(iterDisposable) {
-            _activeCount += 1
             let iter = MergeSinkIter(parent: self, disposeKey: disposeKey)
             let subscription = source.subscribe(iter)
             iterDisposable.setDisposable(subscription)
         }
     }
 
-    func run(_ sources: [SourceElement]) -> Disposable {
-        let _ = _group.insert(_sourceSubscription)
+    func run(_ sources: [Observable<Observer.E>]) -> Disposable {
+        _activeCount += sources.count
 
         for source in sources {
-            self.on(.next(source))
+            subscribeInner(source)
         }
 
         _stopped = true

diff --git a/Sources/AllTestz/main.swift b/Sources/AllTestz/main.swift
@@ -56,6 +56,7 @@ final class AnomaliesTest_ : AnomaliesTest, RxTestCase {
     static var allTests: [(String, (AnomaliesTest_) -> () -> ())] { return [
     ("test936", AnomaliesTest.test936),
     ("test1323", AnomaliesTest.test1323),
+    ("test1344", AnomaliesTest.test1344),
     ("testSeparationBetweenOnAndSubscriptionLocks", AnomaliesTest.testSeparationBetweenOnAndSubscriptionLocks),
     ] }
 }

diff --git a/Tests/RxSwiftTests/Anomalies.swift b/Tests/RxSwiftTests/Anomalies.swift
@@ -101,6 +101,30 @@ extension AnomaliesTest {
         }
     }
 
+    func test1344(){
+        let disposeBag = DisposeBag()
+        let foo = Observable<Int>.create({ observer in
+                observer.on(.next(1))
+                Thread.sleep(forTimeInterval: 0.1)
+                observer.on(.completed)
+                return Disposables.create()
+            })
+            .flatMap { (int) -> Observable<[Int]> in
+                return Observable.create { (observer) -> Disposable in
+                    DispatchQueue.global().async {
+                        observer.onNext([int])
+                    }
+                    self.sleep(0.1)
+                    return Disposables.create()
+                }
+            }
+
+        Observable.merge(foo, .just([42]))
+            .subscribe { (e) in
+            }
+            .disposed(by: disposeBag)
+    }
+
     func testSeparationBetweenOnAndSubscriptionLocks() {
         func performSharingOperatorsTest(share: @escaping (Observable<Int>) -> Observable<Int>) {
             for i in 0 ..< 1 {