Skip to content

Commit ad1840b

Browse files
akarnokdakarnokd
authored
3.x: Add onErrorComplete to Flowable, Observable and Single (#6867)
* 3.x: Add onErrorComplete to F/O/S * Add version tags.
1 parent 13473da commit ad1840b

File tree

11 files changed

+793
-4
lines changed

11 files changed

+793
-4
lines changed

‎src/main/java/io/reactivex/rxjava3/core/Flowable.java

+53-1
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import org.reactivestreams.*;
2020

2121
import io.reactivex.rxjava3.annotations.*;
22+
import io.reactivex.rxjava3.core.Observable;
2223
import io.reactivex.rxjava3.disposables.Disposable;
2324
import io.reactivex.rxjava3.exceptions.*;
2425
import io.reactivex.rxjava3.flowables.*;
@@ -28,7 +29,7 @@
2829
import io.reactivex.rxjava3.internal.jdk8.*;
2930
import io.reactivex.rxjava3.internal.operators.flowable.*;
3031
import io.reactivex.rxjava3.internal.operators.mixed.*;
31-
import io.reactivex.rxjava3.internal.operators.observable.ObservableFromPublisher;
32+
import io.reactivex.rxjava3.internal.operators.observable.*;
3233
import io.reactivex.rxjava3.internal.schedulers.ImmediateThinScheduler;
3334
import io.reactivex.rxjava3.internal.subscribers.*;
3435
import io.reactivex.rxjava3.internal.util.*;
@@ -12308,6 +12309,57 @@ public final Flowable<T> onBackpressureLatest() {
1230812309
return RxJavaPlugins.onAssembly(new FlowableOnBackpressureLatest<>(this));
1230912310
}
1231012311

12312+
/**
12313+
* Returns a {@code Flowable} instance that if the current {@code Flowable} emits an error, it will emit an {@code onComplete}
12314+
* and swallow the throwable.
12315+
* <p>
12316+
* <img width="640" height="372" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Flowable.onErrorComplete.png" alt="">
12317+
* <dl>
12318+
* <dt><b>Backpressure:</b></dt>
12319+
* <dd>The operator doesn't interfere with backpressure which is determined by the current {@code Flowable}'s backpressure
12320+
* behavior.</dd>
12321+
* <dt><b>Scheduler:</b></dt>
12322+
* <dd>{@code onErrorComplete} does not operate by default on a particular {@link Scheduler}.</dd>
12323+
* </dl>
12324+
* @return the new {@code Flowable} instance
12325+
* @since 3.0.0
12326+
*/
12327+
@CheckReturnValue
12328+
@SchedulerSupport(SchedulerSupport.NONE)
12329+
@BackpressureSupport(BackpressureKind.PASS_THROUGH)
12330+
@NonNull
12331+
public final Flowable<T> onErrorComplete() {
12332+
return onErrorComplete(Functions.alwaysTrue());
12333+
}
12334+
12335+
/**
12336+
* Returns a {@code Flowable} instance that if the current {@code Flowable} emits an error and the predicate returns
12337+
* {@code true}, it will emit an {@code onComplete} and swallow the throwable.
12338+
* <p>
12339+
* <img width="640" height="201" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Flowable.onErrorComplete.f.png" alt="">
12340+
* <dl>
12341+
* <dt><b>Backpressure:</b></dt>
12342+
* <dd>The operator doesn't interfere with backpressure which is determined by the current {@code Flowable}'s backpressure
12343+
* behavior.</dd>
12344+
* <dt><b>Scheduler:</b></dt>
12345+
* <dd>{@code onErrorComplete} does not operate by default on a particular {@link Scheduler}.</dd>
12346+
* </dl>
12347+
* @param predicate the predicate to call when an {@link Throwable} is emitted which should return {@code true}
12348+
* if the {@code Throwable} should be swallowed and replaced with an {@code onComplete}.
12349+
* @return the new {@code Flowable} instance
12350+
* @throws NullPointerException if {@code predicate} is {@code null}
12351+
* @since 3.0.0
12352+
*/
12353+
@CheckReturnValue
12354+
@BackpressureSupport(BackpressureKind.PASS_THROUGH)
12355+
@SchedulerSupport(SchedulerSupport.NONE)
12356+
@NonNull
12357+
public final Flowable<T> onErrorComplete(@NonNull Predicate<? super Throwable> predicate) {
12358+
Objects.requireNonNull(predicate, "predicate is null");
12359+
12360+
return RxJavaPlugins.onAssembly(new FlowableOnErrorComplete<>(this, predicate));
12361+
}
12362+
1231112363
/**
1231212364
* Resumes the flow with a {@link Publisher} returned for the failure {@link Throwable} of the current {@code Flowable} by a
1231312365
* function instead of signaling the error via {@code onError}.

‎src/main/java/io/reactivex/rxjava3/core/Maybe.java

+4
Original file line numberDiff line numberDiff line change
@@ -4042,6 +4042,8 @@ public final Single<T> toSingle() {
40424042
/**
40434043
* Returns a {@code Maybe} instance that if this {@code Maybe} emits an error, it will emit an {@code onComplete}
40444044
* and swallow the throwable.
4045+
* <p>
4046+
* <img width="640" height="372" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.onErrorComplete.png" alt="">
40454047
* <dl>
40464048
* <dt><b>Scheduler:</b></dt>
40474049
* <dd>{@code onErrorComplete} does not operate by default on a particular {@link Scheduler}.</dd>
@@ -4058,6 +4060,8 @@ public final Maybe<T> onErrorComplete() {
40584060
/**
40594061
* Returns a {@code Maybe} instance that if this {@code Maybe} emits an error and the predicate returns
40604062
* {@code true}, it will emit an {@code onComplete} and swallow the throwable.
4063+
* <p>
4064+
* <img width="640" height="220" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.onErrorComplete.f.png" alt="">
40614065
* <dl>
40624066
* <dt><b>Scheduler:</b></dt>
40634067
* <dd>{@code onErrorComplete} does not operate by default on a particular {@link Scheduler}.</dd>

‎src/main/java/io/reactivex/rxjava3/core/Observable.java

+43
Original file line numberDiff line numberDiff line change
@@ -10346,6 +10346,49 @@ public final <U> Observable<U> ofType(@NonNull Class<U> clazz) {
1034610346
return filter(Functions.isInstanceOf(clazz)).cast(clazz);
1034710347
}
1034810348

10349+
/**
10350+
* Returns an {@code Observable} instance that if the current {@code Observable} emits an error, it will emit an {@code onComplete}
10351+
* and swallow the throwable.
10352+
* <p>
10353+
* <img width="640" height="373" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Observable.onErrorComplete.png" alt="">
10354+
* <dl>
10355+
* <dt><b>Scheduler:</b></dt>
10356+
* <dd>{@code onErrorComplete} does not operate by default on a particular {@link Scheduler}.</dd>
10357+
* </dl>
10358+
* @return the new {@code Observable} instance
10359+
* @since 3.0.0
10360+
*/
10361+
@CheckReturnValue
10362+
@SchedulerSupport(SchedulerSupport.NONE)
10363+
@NonNull
10364+
public final Observable<T> onErrorComplete() {
10365+
return onErrorComplete(Functions.alwaysTrue());
10366+
}
10367+
10368+
/**
10369+
* Returns an {@code Observable} instance that if the current {@code Observable} emits an error and the predicate returns
10370+
* {@code true}, it will emit an {@code onComplete} and swallow the throwable.
10371+
* <p>
10372+
* <img width="640" height="215" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Observable.onErrorComplete.f.png" alt="">
10373+
* <dl>
10374+
* <dt><b>Scheduler:</b></dt>
10375+
* <dd>{@code onErrorComplete} does not operate by default on a particular {@link Scheduler}.</dd>
10376+
* </dl>
10377+
* @param predicate the predicate to call when an {@link Throwable} is emitted which should return {@code true}
10378+
* if the {@code Throwable} should be swallowed and replaced with an {@code onComplete}.
10379+
* @return the new {@code Observable} instance
10380+
* @throws NullPointerException if {@code predicate} is {@code null}
10381+
* @since 3.0.0
10382+
*/
10383+
@CheckReturnValue
10384+
@NonNull
10385+
@SchedulerSupport(SchedulerSupport.NONE)
10386+
public final Observable<T> onErrorComplete(@NonNull Predicate<? super Throwable> predicate) {
10387+
Objects.requireNonNull(predicate, "predicate is null");
10388+
10389+
return RxJavaPlugins.onAssembly(new ObservableOnErrorComplete<>(this, predicate));
10390+
}
10391+
1034910392
/**
1035010393
* Resumes the flow with an {@link ObservableSource} returned for the failure {@link Throwable} of the current {@code Observable} by a
1035110394
* function instead of signaling the error via {@code onError}.

‎src/main/java/io/reactivex/rxjava3/core/Single.java

+43
Original file line numberDiff line numberDiff line change
@@ -3439,6 +3439,49 @@ public final Single<T> onErrorResumeWith(@NonNull SingleSource<? extends T> fall
34393439
return onErrorResumeNext(Functions.justFunction(fallback));
34403440
}
34413441

3442+
/**
3443+
* Returns a {@link Maybe} instance that if the current {@code Single} emits an error, it will emit an {@code onComplete}
3444+
* and swallow the throwable.
3445+
* <p>
3446+
* <img width="640" height="554" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.onErrorComplete.png" alt="">
3447+
* <dl>
3448+
* <dt><b>Scheduler:</b></dt>
3449+
* <dd>{@code onErrorComplete} does not operate by default on a particular {@link Scheduler}.</dd>
3450+
* </dl>
3451+
* @return the new {@code Maybe} instance
3452+
* @since 3.0.0
3453+
*/
3454+
@CheckReturnValue
3455+
@SchedulerSupport(SchedulerSupport.NONE)
3456+
@NonNull
3457+
public final Maybe<T> onErrorComplete() {
3458+
return onErrorComplete(Functions.alwaysTrue());
3459+
}
3460+
3461+
/**
3462+
* Returns a {@link Maybe} instance that if this {@code Single} emits an error and the predicate returns
3463+
* {@code true}, it will emit an {@code onComplete} and swallow the throwable.
3464+
* <p>
3465+
* <img width="640" height="220" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.onErrorComplete.f.png" alt="">
3466+
* <dl>
3467+
* <dt><b>Scheduler:</b></dt>
3468+
* <dd>{@code onErrorComplete} does not operate by default on a particular {@link Scheduler}.</dd>
3469+
* </dl>
3470+
* @param predicate the predicate to call when an {@link Throwable} is emitted which should return {@code true}
3471+
* if the {@code Throwable} should be swallowed and replaced with an {@code onComplete}.
3472+
* @return the new {@code Maybe} instance
3473+
* @throws NullPointerException if {@code predicate} is {@code null}
3474+
* @since 3.0.0
3475+
*/
3476+
@CheckReturnValue
3477+
@NonNull
3478+
@SchedulerSupport(SchedulerSupport.NONE)
3479+
public final Maybe<T> onErrorComplete(@NonNull Predicate<? super Throwable> predicate) {
3480+
Objects.requireNonNull(predicate, "predicate is null");
3481+
3482+
return RxJavaPlugins.onAssembly(new SingleOnErrorComplete<>(this, predicate));
3483+
}
3484+
34423485
/**
34433486
* Resumes the flow with a {@link SingleSource} returned for the failure {@link Throwable} of the current {@code Single} by a
34443487
* function instead of signaling the error via {@code onError}.

‎src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnErrorComplete.java

+107
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
/**
2+
* Copyright (c) 2016-present, RxJava Contributors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
5+
* compliance with the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is
10+
* distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
11+
* the License for the specific language governing permissions and limitations under the License.
12+
*/
13+
14+
package io.reactivex.rxjava3.internal.operators.flowable;
15+
16+
import org.reactivestreams.*;
17+
18+
import io.reactivex.rxjava3.core.*;
19+
import io.reactivex.rxjava3.exceptions.*;
20+
import io.reactivex.rxjava3.functions.Predicate;
21+
import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper;
22+
23+
/**
24+
* Emits an onComplete if the source emits an onError and the predicate returns true for
25+
* that Throwable.
26+
*
27+
* @param <T> the value type
28+
* @since 3.0.0
29+
*/
30+
public final class FlowableOnErrorComplete<T> extends AbstractFlowableWithUpstream<T, T> {
31+
32+
final Predicate<? super Throwable> predicate;
33+
34+
public FlowableOnErrorComplete(Flowable<T> source,
35+
Predicate<? super Throwable> predicate) {
36+
super(source);
37+
this.predicate = predicate;
38+
}
39+
40+
@Override
41+
protected void subscribeActual(Subscriber<? super T> observer) {
42+
source.subscribe(new OnErrorCompleteSubscriber<>(observer, predicate));
43+
}
44+
45+
public static final class OnErrorCompleteSubscriber<T>
46+
implements FlowableSubscriber<T>, Subscription {
47+
48+
final Subscriber<? super T> downstream;
49+
50+
final Predicate<? super Throwable> predicate;
51+
52+
Subscription upstream;
53+
54+
public OnErrorCompleteSubscriber(Subscriber<? super T> actual, Predicate<? super Throwable> predicate) {
55+
this.downstream = actual;
56+
this.predicate = predicate;
57+
}
58+
59+
@Override
60+
public void onSubscribe(Subscription s) {
61+
if (SubscriptionHelper.validate(this.upstream, s)) {
62+
this.upstream = s;
63+
64+
downstream.onSubscribe(this);
65+
}
66+
}
67+
68+
@Override
69+
public void onNext(T value) {
70+
downstream.onNext(value);
71+
}
72+
73+
@Override
74+
public void onError(Throwable e) {
75+
boolean b;
76+
77+
try {
78+
b = predicate.test(e);
79+
} catch (Throwable ex) {
80+
Exceptions.throwIfFatal(ex);
81+
downstream.onError(new CompositeException(e, ex));
82+
return;
83+
}
84+
85+
if (b) {
86+
downstream.onComplete();
87+
} else {
88+
downstream.onError(e);
89+
}
90+
}
91+
92+
@Override
93+
public void onComplete() {
94+
downstream.onComplete();
95+
}
96+
97+
@Override
98+
public void cancel() {
99+
upstream.cancel();
100+
}
101+
102+
@Override
103+
public void request(long n) {
104+
upstream.request(n);
105+
}
106+
}
107+
}

‎src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeOnErrorComplete.java

+4-3
Original file line numberDiff line numberDiff line change
@@ -37,18 +37,19 @@ public MaybeOnErrorComplete(MaybeSource<T> source,
3737

3838
@Override
3939
protected void subscribeActual(MaybeObserver<? super T> observer) {
40-
source.subscribe(new OnErrorCompleteMaybeObserver<>(observer, predicate));
40+
source.subscribe(new OnErrorCompleteMultiObserver<>(observer, predicate));
4141
}
4242

43-
static final class OnErrorCompleteMaybeObserver<T> implements MaybeObserver<T>, Disposable {
43+
public static final class OnErrorCompleteMultiObserver<T>
44+
implements MaybeObserver<T>, SingleObserver<T>, Disposable {
4445

4546
final MaybeObserver<? super T> downstream;
4647

4748
final Predicate<? super Throwable> predicate;
4849

4950
Disposable upstream;
5051

51-
OnErrorCompleteMaybeObserver(MaybeObserver<? super T> actual, Predicate<? super Throwable> predicate) {
52+
public OnErrorCompleteMultiObserver(MaybeObserver<? super T> actual, Predicate<? super Throwable> predicate) {
5253
this.downstream = actual;
5354
this.predicate = predicate;
5455
}

0 commit comments

Comments
 (0)