Fixes issues with andThen operator. #1347

ReactiveX · Jul 20, 2017 · eb25e70 · eb25e70
1 parent b343881
commit eb25e70
Show file tree

Hide file tree

Showing 3 changed files with 162 additions and 1 deletion.
diff --git a/RxSwift/Traits/Completable+AndThen.swift b/RxSwift/Traits/Completable+AndThen.swift
@@ -94,6 +94,7 @@ final fileprivate class ConcatCompletableSink<O: ObserverType>
         switch event {
         case .error(let error):
             self.forwardOn(.error(error))
+            self.dispose()
         case .next:
             break
         case .completed:
@@ -103,7 +104,9 @@ final fileprivate class ConcatCompletableSink<O: ObserverType>
     }
 
     func run() -> Disposable {
-        _subscription.disposable = _parent._completable.subscribe(self)
+        let subscription = SingleAssignmentDisposable()
+        _subscription.disposable = subscription
+        subscription.setDisposable(_parent._completable.subscribe(self))
         return _subscription
     }
 }
@@ -122,5 +125,8 @@ final fileprivate class ConcatCompletableSinkOther<O: ObserverType>
 
     func on(_ event: Event<O.E>) {
         _parent.forwardOn(event)
+        if event.isStopEvent {
+            _parent.dispose()
+        }
     }
 }
diff --git a/Sources/AllTestz/main.swift b/Sources/AllTestz/main.swift
@@ -470,16 +470,20 @@ final class CompletableAndThenTest_ : CompletableAndThenTest, RxTestCase {
     #endif
 
     static var allTests: [(String, (CompletableAndThenTest_) -> () -> ())] { return [
+    ("testCompletableEmpty_CompletableCompleted", CompletableAndThenTest.testCompletableEmpty_CompletableCompleted),
     ("testCompletableCompleted_CompletableCompleted", CompletableAndThenTest.testCompletableCompleted_CompletableCompleted),
     ("testCompletableError_CompletableCompleted", CompletableAndThenTest.testCompletableError_CompletableCompleted),
     ("testCompletableCompleted_CompletableError", CompletableAndThenTest.testCompletableCompleted_CompletableError),
+    ("testCompletableEmpty_SingleCompleted", CompletableAndThenTest.testCompletableEmpty_SingleCompleted),
     ("testCompletableCompleted_SingleNormal", CompletableAndThenTest.testCompletableCompleted_SingleNormal),
     ("testCompletableError_SingleNormal", CompletableAndThenTest.testCompletableError_SingleNormal),
     ("testCompletableCompleted_SingleError", CompletableAndThenTest.testCompletableCompleted_SingleError),
+    ("testCompletableEmpty_MaybeCompleted", CompletableAndThenTest.testCompletableEmpty_MaybeCompleted),
     ("testCompletableCompleted_MaybeNormal", CompletableAndThenTest.testCompletableCompleted_MaybeNormal),
     ("testCompletableError_MaybeNormal", CompletableAndThenTest.testCompletableError_MaybeNormal),
     ("testCompletableCompleted_MaybeError", CompletableAndThenTest.testCompletableCompleted_MaybeError),
     ("testCompletableCompleted_MaybeEmpty", CompletableAndThenTest.testCompletableCompleted_MaybeEmpty),
+    ("testCompletableEmpty_ObservableCompleted", CompletableAndThenTest.testCompletableEmpty_ObservableCompleted),
     ("testCompletableCompleted_ObservableNormal", CompletableAndThenTest.testCompletableCompleted_ObservableNormal),
     ("testCompletableError_ObservableNormal", CompletableAndThenTest.testCompletableError_ObservableNormal),
     ("testCompletableCompleted_ObservableError", CompletableAndThenTest.testCompletableCompleted_ObservableError),

diff --git a/Tests/RxSwiftTests/Completable+AndThen.swift b/Tests/RxSwiftTests/Completable+AndThen.swift
@@ -15,6 +15,28 @@ class CompletableAndThenTest : RxTest {
 
 extension CompletableAndThenTest {
 
+    func testCompletableEmpty_CompletableCompleted() {
+        let scheduler = TestScheduler(initialClock: 0)
+
+        let x1: Completable = Completable.empty()
+
+        let x2: TestableObservable<Never> = scheduler.createHotObservable([
+            completed(290),
+            ])
+
+        let res = scheduler.start {
+            x1.andThen(x2.asCompletable()).asObservable()
+        }
+
+        XCTAssertEqual(res.events, [
+            completed(290)
+            ])
+
+        XCTAssertEqual(x2.subscriptions, [
+            Subscription(200, 290)
+            ])
+    }
+
     func testCompletableCompleted_CompletableCompleted() {
         let scheduler = TestScheduler(initialClock: 0)
 
@@ -97,10 +119,48 @@ extension CompletableAndThenTest {
             Subscription(210, 290)
             ])
     }
+
+    #if TRACE_RESOURCES
+        func testAndThenCompletableReleasesResourcesOnComplete() {
+            _ = Completable.empty().andThen(Completable.empty()).subscribe()
+        }
+
+        func testAndThenCompletableReleasesResourcesOnError1() {
+            _ = Completable.error(testError).andThen(Completable.empty()).subscribe()
+        }
+
+        func testAndThenCompletableReleasesResourcesOnError2() {
+            _ = Completable.empty().andThen(Completable.error(testError)).subscribe()
+        }
+    #endif
 }
 
 extension CompletableAndThenTest {
 
+    func testCompletableEmpty_SingleCompleted() {
+        let scheduler = TestScheduler(initialClock: 0)
+
+        let x1: Completable = Completable.empty()
+
+        let x2: TestableObservable<Int> = scheduler.createHotObservable([
+            next(285, 1),
+            completed(290),
+            ])
+
+        let res = scheduler.start {
+            x1.andThen(x2.asSingle()).asObservable()
+        }
+
+        XCTAssertEqual(res.events, [
+            next(290, 1),
+            completed(290)
+            ])
+
+        XCTAssertEqual(x2.subscriptions, [
+            Subscription(200, 290)
+            ])
+    }
+
     func testCompletableCompleted_SingleNormal() {
         let scheduler = TestScheduler(initialClock: 0)
 
@@ -187,10 +247,48 @@ extension CompletableAndThenTest {
             Subscription(210, 290)
             ])
     }
+
+    #if TRACE_RESOURCES
+        func testAndThenSingleReleasesResourcesOnComplete() {
+            _ = Completable.empty().andThen(Single.just(1)).subscribe()
+        }
+
+        func testAndThenSingleReleasesResourcesOnError1() {
+            _ = Completable.error(testError).andThen(Single.just(1)).subscribe()
+        }
+
+        func testAndThenSingleReleasesResourcesOnError2() {
+            _ = Completable.empty().andThen(Single<Int>.error(testError)).subscribe()
+        }
+    #endif
 }
 
 extension CompletableAndThenTest {
 
+    func testCompletableEmpty_MaybeCompleted() {
+        let scheduler = TestScheduler(initialClock: 0)
+
+        let x1: Completable = Completable.empty()
+
+        let x2: TestableObservable<Int> = scheduler.createHotObservable([
+            next(285, 1),
+            completed(290),
+            ])
+
+        let res = scheduler.start {
+            x1.andThen(x2.asMaybe()).asObservable()
+        }
+
+        XCTAssertEqual(res.events, [
+            next(290, 1),
+            completed(290)
+            ])
+
+        XCTAssertEqual(x2.subscriptions, [
+            Subscription(200, 290)
+            ])
+    }
+
     func testCompletableCompleted_MaybeNormal() {
         let scheduler = TestScheduler(initialClock: 0)
 
@@ -305,10 +403,48 @@ extension CompletableAndThenTest {
             Subscription(210, 290)
             ])
     }
+
+    #if TRACE_RESOURCES
+        func testAndThenMaybeReleasesResourcesOnComplete() {
+            _ = Completable.empty().andThen(Maybe.just(1)).subscribe()
+        }
+
+        func testAndThenMaybeReleasesResourcesOnError1() {
+            _ = Completable.error(testError).andThen(Maybe.just(1)).subscribe()
+        }
+
+        func testAndThenMaybeReleasesResourcesOnError2() {
+            _ = Completable.empty().andThen(Maybe<Int>.error(testError)).subscribe()
+        }
+    #endif
 }
 
 extension CompletableAndThenTest {
 
+    func testCompletableEmpty_ObservableCompleted() {
+        let scheduler = TestScheduler(initialClock: 0)
+
+        let x1: Completable = Completable.empty()
+
+        let x2: TestableObservable<Int> = scheduler.createHotObservable([
+            next(285, 1),
+            completed(290),
+            ])
+
+        let res = scheduler.start {
+            x1.andThen(x2.asObservable()).asObservable()
+        }
+
+        XCTAssertEqual(res.events, [
+            next(285, 1),
+            completed(290)
+            ])
+
+        XCTAssertEqual(x2.subscriptions, [
+            Subscription(200, 290)
+            ])
+    }
+
     func testCompletableCompleted_ObservableNormal() {
         let scheduler = TestScheduler(initialClock: 0)
 
@@ -423,4 +559,19 @@ extension CompletableAndThenTest {
             Subscription(210, 290)
             ])
     }
+
+
+    #if TRACE_RESOURCES
+        func testAndThenObservableReleasesResourcesOnComplete() {
+            _ = Completable.empty().andThen(Observable.just(1)).subscribe()
+        }
+
+        func testAndThenObservableReleasesResourcesOnError1() {
+            _ = Completable.error(testError).andThen(Observable.just(1)).subscribe()
+        }
+
+        func testAndThenObservableReleasesResourcesOnError2() {
+            _ = Completable.empty().andThen(Observable<Int>.error(testError)).subscribe()
+        }
+    #endif
 }