Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

2.x: Fix Observable.switchMap main onError not disposing the current inner source #5833

Merged
merged 2 commits into from
Feb 2, 2018
Merged

2.x: Fix Observable.switchMap main onError not disposing the current inner source #5833

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8ec030f
2.x: Fix Obs.switchMap main onError not disposing the current inner src
akarnokd Feb 1, 2018
cfa17ae
Fix error-error race test
akarnokd Feb 2, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions src/main/java/io/reactivex/internal/operators/observable/ObservableSwitchMap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,15 +131,15 @@ public void onNext(T t) {

@Override
public void onError(Throwable t) {
if (done || !errors.addThrowable(t)) {
if (!done && errors.addThrowable(t)) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Codecov indicates that some of possible variants are not tested, do you want to cover them in this PR?

screen shot 2018-02-01 at 4 48 30 pm

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's not that I think code in the patch is broken, but that if we had test that was trying to go into onError when it's already done == true, we would have a hint that original code had some problems

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That line is tested but parallelism on the CI is flaky and may not produce the necessary interleaving there.

The upstream may onError after a crash in the mapper thus we have to route that into the global error handler.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can't we test it just by breaking RS protocol with main.onComplete(); main.onError()?

I might miss something, reviewing on mobile

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This and this tests should cover that line. Apparently, the test did not involve ps2 so the inner source was not involved. Updated the unit tests accordingly.

if (!delayErrors) {
disposeInner();
}
done = true;
drain();
} else {
RxJavaPlugins.onError(t);
return;
}
done = true;
drain();
}

@Override
Expand Down
21 changes: 21 additions & 0 deletions src/test/java/io/reactivex/internal/operators/flowable/FlowableSwitchTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1153,4 +1153,25 @@ public Object apply(Integer v) throws Exception {
.test()
.assertFailure(TestException.class);
}

@Test
public void innerCancelledOnMainError() {
final PublishProcessor<Integer> main = PublishProcessor.create();
final PublishProcessor<Integer> inner = PublishProcessor.create();

TestSubscriber<Integer> to = main.switchMap(Functions.justFunction(inner))
.test();

assertTrue(main.hasSubscribers());

main.onNext(1);

assertTrue(inner.hasSubscribers());

main.onError(new TestException());

assertFalse(inner.hasSubscribers());

to.assertFailure(TestException.class);
}
}
21 changes: 21 additions & 0 deletions src/test/java/io/reactivex/internal/operators/observable/ObservableSwitchTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -963,4 +963,25 @@ public void onNext(Integer t) {

to.assertFailure(TestException.class, 1);
}

@Test
public void innerDisposedOnMainError() {
final PublishSubject<Integer> main = PublishSubject.create();
final PublishSubject<Integer> inner = PublishSubject.create();

TestObserver<Integer> to = main.switchMap(Functions.justFunction(inner))
.test();

assertTrue(main.hasObservers());

main.onNext(1);

assertTrue(inner.hasObservers());

main.onError(new TestException());

assertFalse(inner.hasObservers());

to.assertFailure(TestException.class);
}
}