Skip to content

3.x: Fix many operators swallowing undeliverable exceptions #6612

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

Merged
merged 2 commits into from
Aug 5, 2019
Merged

3.x: Fix many operators swallowing undeliverable exceptions #6612

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1300964
3.x: Fix many operators swallowing undeliverable exceptions
akarnokd Aug 5, 2019
822145d
Add Completable.merge variants too
akarnokd Aug 5, 2019
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
23 changes: 7 additions & 16 deletions src/main/java/io/reactivex/internal/operators/completable/CompletableMerge.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@ static final class CompletableMergeSubscriber
public void dispose() {
upstream.cancel();
set.dispose();
error.tryTerminateAndReport();
}

@Override
Expand Down Expand Up @@ -106,15 +107,15 @@ public void onError(Throwable t) {

if (error.addThrowable(t)) {
if (getAndSet(0) > 0) {
downstream.onError(error.terminate());
error.tryTerminateConsumer(downstream);
}
} else {
RxJavaPlugins.onError(t);
}
} else {
if (error.addThrowable(t)) {
if (decrementAndGet() == 0) {
downstream.onError(error.terminate());
error.tryTerminateConsumer(downstream);
}
} else {
RxJavaPlugins.onError(t);
Expand All @@ -125,12 +126,7 @@ public void onError(Throwable t) {
@Override
public void onComplete() {
if (decrementAndGet() == 0) {
Throwable ex = error.get();
if (ex != null) {
downstream.onError(error.terminate());
} else {
downstream.onComplete();
}
error.tryTerminateConsumer(downstream);
}
}

Expand All @@ -142,15 +138,15 @@ void innerError(MergeInnerObserver inner, Throwable t) {

if (error.addThrowable(t)) {
if (getAndSet(0) > 0) {
downstream.onError(error.terminate());
error.tryTerminateConsumer(downstream);
}
} else {
RxJavaPlugins.onError(t);
}
} else {
if (error.addThrowable(t)) {
if (decrementAndGet() == 0) {
downstream.onError(error.terminate());
error.tryTerminateConsumer(downstream);
} else {
if (maxConcurrency != Integer.MAX_VALUE) {
upstream.request(1);
Expand All @@ -165,12 +161,7 @@ void innerError(MergeInnerObserver inner, Throwable t) {
void innerComplete(MergeInnerObserver inner) {
set.delete(inner);
if (decrementAndGet() == 0) {
Throwable ex = error.get();
if (ex != null) {
downstream.onError(ex);
} else {
downstream.onComplete();
}
error.tryTerminateConsumer(downstream);
} else {
if (maxConcurrency != Integer.MAX_VALUE) {
upstream.request(1);
Expand Down
15 changes: 13 additions & 2 deletions src/main/java/io/reactivex/internal/operators/completable/CompletableMergeArray.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ public void subscribeActual(final CompletableObserver observer) {
final AtomicBoolean once = new AtomicBoolean();

InnerCompletableObserver shared = new InnerCompletableObserver(observer, once, set, sources.length + 1);
observer.onSubscribe(set);
observer.onSubscribe(shared);

for (CompletableSource c : sources) {
if (set.isDisposed()) {
Expand All @@ -52,7 +52,7 @@ public void subscribeActual(final CompletableObserver observer) {
shared.onComplete();
}

static final class InnerCompletableObserver extends AtomicInteger implements CompletableObserver {
static final class InnerCompletableObserver extends AtomicInteger implements CompletableObserver, Disposable {
private static final long serialVersionUID = -8360547806504310570L;

final CompletableObserver downstream;
Expand Down Expand Up @@ -91,5 +91,16 @@ public void onComplete() {
}
}
}

@Override
public void dispose() {
set.dispose();
once.set(true);
}

@Override
public boolean isDisposed() {
return set.isDisposed();
}
}
}
32 changes: 20 additions & 12 deletions ...ain/java/io/reactivex/internal/operators/completable/CompletableMergeDelayErrorArray.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ public void subscribeActual(final CompletableObserver observer) {
final AtomicInteger wip = new AtomicInteger(sources.length + 1);

final AtomicThrowable error = new AtomicThrowable();
set.add(new TryTerminateAndReportDisposable(error));

observer.onSubscribe(set);

Expand All @@ -53,12 +54,24 @@ public void subscribeActual(final CompletableObserver observer) {
}

if (wip.decrementAndGet() == 0) {
Throwable ex = error.terminate();
if (ex == null) {
observer.onComplete();
} else {
observer.onError(ex);
}
error.tryTerminateConsumer(observer);
}
}

static final class TryTerminateAndReportDisposable implements Disposable {
final AtomicThrowable errors;
TryTerminateAndReportDisposable(AtomicThrowable errors) {
this.errors = errors;
}

@Override
public void dispose() {
errors.tryTerminateAndReport();
}

@Override
public boolean isDisposed() {
return errors.isTerminated();
}
}

Expand Down Expand Up @@ -98,12 +111,7 @@ public void onComplete() {

void tryTerminate() {
if (wip.decrementAndGet() == 0) {
Throwable ex = error.terminate();
if (ex == null) {
downstream.onComplete();
} else {
downstream.onError(ex);
}
error.tryTerminateConsumer(downstream);
}
}
}
Expand Down
10 changes: 3 additions & 7 deletions .../java/io/reactivex/internal/operators/completable/CompletableMergeDelayErrorIterable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.exceptions.Exceptions;
import io.reactivex.internal.functions.ObjectHelper;
import io.reactivex.internal.operators.completable.CompletableMergeDelayErrorArray.MergeInnerCompletableObserver;
import io.reactivex.internal.operators.completable.CompletableMergeDelayErrorArray.*;
import io.reactivex.internal.util.AtomicThrowable;

public final class CompletableMergeDelayErrorIterable extends Completable {
Expand Down Expand Up @@ -50,6 +50,7 @@ public void subscribeActual(final CompletableObserver observer) {
final AtomicInteger wip = new AtomicInteger(1);

final AtomicThrowable error = new AtomicThrowable();
set.add(new TryTerminateAndReportDisposable(error));

for (;;) {
if (set.isDisposed()) {
Expand Down Expand Up @@ -93,12 +94,7 @@ public void subscribeActual(final CompletableObserver observer) {
}

if (wip.decrementAndGet() == 0) {
Throwable ex = error.terminate();
if (ex == null) {
observer.onComplete();
} else {
observer.onError(ex);
}
error.tryTerminateConsumer(observer);
}
}
}
21 changes: 16 additions & 5 deletions src/main/java/io/reactivex/internal/operators/completable/CompletableMergeIterable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,8 +32,11 @@ public CompletableMergeIterable(Iterable<? extends CompletableSource> sources) {
@Override
public void subscribeActual(final CompletableObserver observer) {
final CompositeDisposable set = new CompositeDisposable();
final AtomicInteger wip = new AtomicInteger(1);

MergeCompletableObserver shared = new MergeCompletableObserver(observer, set, wip);

observer.onSubscribe(set);
observer.onSubscribe(shared);

Iterator<? extends CompletableSource> iterator;

Expand All @@ -45,9 +48,6 @@ public void subscribeActual(final CompletableObserver observer) {
return;
}

final AtomicInteger wip = new AtomicInteger(1);

MergeCompletableObserver shared = new MergeCompletableObserver(observer, set, wip);
for (;;) {
if (set.isDisposed()) {
return;
Expand Down Expand Up @@ -94,7 +94,7 @@ public void subscribeActual(final CompletableObserver observer) {
shared.onComplete();
}

static final class MergeCompletableObserver extends AtomicBoolean implements CompletableObserver {
static final class MergeCompletableObserver extends AtomicBoolean implements CompletableObserver, Disposable {

private static final long serialVersionUID = -7730517613164279224L;

Expand Down Expand Up @@ -133,5 +133,16 @@ public void onComplete() {
}
}
}

@Override
public void dispose() {
set.dispose();
set(true);
}

@Override
public boolean isDisposed() {
return set.isDisposed();
}
}
}
38 changes: 20 additions & 18 deletions src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatMap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,7 +200,7 @@ public void onError(Throwable t) {
inner.cancel();

if (getAndIncrement() == 0) {
downstream.onError(errors.terminate());
errors.tryTerminateConsumer(downstream);
}
} else {
RxJavaPlugins.onError(t);
Expand All @@ -214,7 +214,7 @@ public void innerNext(R value) {
if (compareAndSet(1, 0)) {
return;
}
downstream.onError(errors.terminate());
errors.tryTerminateConsumer(downstream);
}
}

Expand All @@ -224,7 +224,7 @@ public void innerError(Throwable e) {
upstream.cancel();

if (getAndIncrement() == 0) {
downstream.onError(errors.terminate());
errors.tryTerminateConsumer(downstream);
}
} else {
RxJavaPlugins.onError(e);
Expand All @@ -243,6 +243,8 @@ public void cancel() {

inner.cancel();
upstream.cancel();

errors.tryTerminateAndReport();
}
}

Expand All @@ -265,7 +267,7 @@ void drain() {
Exceptions.throwIfFatal(e);
upstream.cancel();
errors.addThrowable(e);
downstream.onError(errors.terminate());
errors.tryTerminateConsumer(downstream);
return;
}

Expand All @@ -286,7 +288,7 @@ void drain() {

upstream.cancel();
errors.addThrowable(e);
downstream.onError(errors.terminate());
errors.tryTerminateConsumer(downstream);
return;
}

Expand All @@ -312,7 +314,7 @@ void drain() {
Exceptions.throwIfFatal(e);
upstream.cancel();
errors.addThrowable(e);
downstream.onError(errors.terminate());
errors.tryTerminateConsumer(downstream);
return;
}

Expand All @@ -324,7 +326,7 @@ void drain() {
if (get() == 0 && compareAndSet(0, 1)) {
downstream.onNext(vr);
if (!compareAndSet(1, 0)) {
downstream.onError(errors.terminate());
errors.tryTerminateConsumer(downstream);
return;
}
}
Expand Down Expand Up @@ -437,6 +439,8 @@ public void cancel() {

inner.cancel();
upstream.cancel();

errors.tryTerminateAndReport();
}
}

Expand All @@ -456,7 +460,7 @@ void drain() {
if (d && !veryEnd) {
Throwable ex = errors.get();
if (ex != null) {
downstream.onError(errors.terminate());
errors.tryTerminateConsumer(downstream);
return;
}
}
Expand All @@ -468,20 +472,18 @@ void drain() {
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
upstream.cancel();
errors.addThrowable(e);
downstream.onError(errors.terminate());
if (errors.addThrowable(e)) {
errors.tryTerminateConsumer(downstream);
} else {
RxJavaPlugins.onError(e);
}
return;
}

boolean empty = v == null;

if (d && empty) {
Throwable ex = errors.terminate();
if (ex != null) {
downstream.onError(ex);
} else {
downstream.onComplete();
}
errors.tryTerminateConsumer(downstream);
return;
}

Expand All @@ -495,7 +497,7 @@ void drain() {

upstream.cancel();
errors.addThrowable(e);
downstream.onError(errors.terminate());
errors.tryTerminateConsumer(downstream);
return;
}

Expand All @@ -522,7 +524,7 @@ void drain() {
errors.addThrowable(e);
if (!veryEnd) {
upstream.cancel();
downstream.onError(errors.terminate());
errors.tryTerminateConsumer(downstream);
return;
}
vr = null;
Expand Down
Loading