Skip to content

Commit ceded86

Browse files
akarnokdakarnokd
authored
2.x: reintroduce OnErrorNotImplementedException for 0-1 arg subscribe() (#5036)
1 parent 7ff1eef commit ceded86

File tree

12 files changed

+267
-26
lines changed

12 files changed

+267
-26
lines changed

‎src/main/java/io/reactivex/Completable.java

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1702,6 +1702,10 @@ public final Disposable subscribe(final Action onComplete, final Consumer<? supe
17021702
* Subscribes to this Completable and calls the given Action when this Completable
17031703
* completes normally.
17041704
* <p>
1705+
* If the Completable emits an error, it is wrapped into an
1706+
* {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
1707+
* and routed to the RxJavaPlugins.onError handler.
1708+
* <p>
17051709
* If this Completable emits an error, it is sent to RxJavaPlugins.onError and gets swallowed.
17061710
* <dl>
17071711
* <dt><b>Scheduler:</b></dt>

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

Lines changed: 19 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5559,6 +5559,11 @@ public final void blockingSubscribe() {
55595559

55605560
/**
55615561
* Subscribes to the source and calls the given callbacks <strong>on the current thread</strong>.
5562+
* <p>
5563+
* If the Flowable emits an error, it is wrapped into an
5564+
* {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
5565+
* and routed to the RxJavaPlugins.onError handler.
5566+
* <p>
55625567
* <dl>
55635568
* <dt><b>Backpressure:</b></dt>
55645569
* <dd>The operator consumes the source {@code Flowable} in an unbounded manner
@@ -5572,7 +5577,7 @@ public final void blockingSubscribe() {
55725577
@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
55735578
@SchedulerSupport(SchedulerSupport.NONE)
55745579
public final void blockingSubscribe(Consumer<? super T> onNext) {
5575-
FlowableBlockingSubscribe.subscribe(this, onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION);
5580+
FlowableBlockingSubscribe.subscribe(this, onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION);
55765581
}
55775582

55785583
/**
@@ -8983,6 +8988,10 @@ public final Disposable forEach(Consumer<? super T> onNext) {
89838988
/**
89848989
* Subscribes to the {@link Publisher} and receives notifications for each element until the
89858990
* onNext Predicate returns false.
8991+
* <p>
8992+
* If the Flowable emits an error, it is wrapped into an
8993+
* {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
8994+
* and routed to the RxJavaPlugins.onError handler.
89868995
* <dl>
89878996
* <dt><b>Backpressure:</b><dt>
89888997
* <dd>The operator consumes the source {@code Publisher} in an unbounded manner (i.e., no
@@ -9003,7 +9012,7 @@ public final Disposable forEach(Consumer<? super T> onNext) {
90039012
@BackpressureSupport(BackpressureKind.NONE)
90049013
@SchedulerSupport(SchedulerSupport.NONE)
90059014
public final Disposable forEachWhile(Predicate<? super T> onNext) {
9006-
return forEachWhile(onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION);
9015+
return forEachWhile(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION);
90079016
}
90089017

90099018
/**
@@ -12713,7 +12722,9 @@ public final Flowable<T> strict() {
1271312722
/**
1271412723
* Subscribes to a Publisher and ignores {@code onNext} and {@code onComplete} emissions.
1271512724
* <p>
12716-
* If the Flowable emits an error, it is routed to the RxJavaPlugins.onError handler.
12725+
* If the Flowable emits an error, it is wrapped into an
12726+
* {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
12727+
* and routed to the RxJavaPlugins.onError handler.
1271712728
* <dl>
1271812729
* <dt><b>Backpressure:</b><dt>
1271912730
* <dd>The operator consumes the source {@code Publisher} in an unbounded manner (i.e., no
@@ -12729,14 +12740,16 @@ public final Flowable<T> strict() {
1272912740
@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
1273012741
@SchedulerSupport(SchedulerSupport.NONE)
1273112742
public final Disposable subscribe() {
12732-
return subscribe(Functions.emptyConsumer(), Functions.ERROR_CONSUMER,
12743+
return subscribe(Functions.emptyConsumer(), Functions.ON_ERROR_MISSING,
1273312744
Functions.EMPTY_ACTION, FlowableInternalHelper.RequestMax.INSTANCE);
1273412745
}
1273512746

1273612747
/**
1273712748
* Subscribes to a Publisher and provides a callback to handle the items it emits.
1273812749
* <p>
12739-
* If the Flowable emits an error, it is routed to the RxJavaPlugins.onError handler.
12750+
* If the Flowable emits an error, it is wrapped into an
12751+
* {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
12752+
* and routed to the RxJavaPlugins.onError handler.
1274012753
* <dl>
1274112754
* <dt><b>Backpressure:</b><dt>
1274212755
* <dd>The operator consumes the source {@code Publisher} in an unbounded manner (i.e., no
@@ -12757,7 +12770,7 @@ public final Disposable subscribe() {
1275712770
@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
1275812771
@SchedulerSupport(SchedulerSupport.NONE)
1275912772
public final Disposable subscribe(Consumer<? super T> onNext) {
12760-
return subscribe(onNext, Functions.ERROR_CONSUMER,
12773+
return subscribe(onNext, Functions.ON_ERROR_MISSING,
1276112774
Functions.EMPTY_ACTION, FlowableInternalHelper.RequestMax.INSTANCE);
1276212775
}
1276312776

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

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3592,7 +3592,9 @@ public final Maybe<T> retryWhen(
35923592
/**
35933593
* Subscribes to a Maybe and ignores {@code onSuccess} and {@code onComplete} emissions.
35943594
* <p>
3595-
* If the Maybe emits an error, it is routed to the RxJavaPlugins.onError handler.
3595+
* If the Maybe emits an error, it is wrapped into an
3596+
* {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
3597+
* and routed to the RxJavaPlugins.onError handler.
35963598
* <dl>
35973599
* <dt><b>Scheduler:</b></dt>
35983600
* <dd>{@code subscribe} does not operate by default on a particular {@link Scheduler}.</dd>
@@ -3604,13 +3606,15 @@ public final Maybe<T> retryWhen(
36043606
*/
36053607
@SchedulerSupport(SchedulerSupport.NONE)
36063608
public final Disposable subscribe() {
3607-
return subscribe(Functions.emptyConsumer(), Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION);
3609+
return subscribe(Functions.emptyConsumer(), Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION);
36083610
}
36093611

36103612
/**
36113613
* Subscribes to a Maybe and provides a callback to handle the items it emits.
36123614
* <p>
3613-
* If the Maybe emits an error, it is routed to the RxJavaPlugins.onError handler.
3615+
* If the Maybe emits an error, it is wrapped into an
3616+
* {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
3617+
* and routed to the RxJavaPlugins.onError handler.
36143618
* <dl>
36153619
* <dt><b>Scheduler:</b></dt>
36163620
* <dd>{@code subscribe} does not operate by default on a particular {@link Scheduler}.</dd>
@@ -3627,7 +3631,7 @@ public final Disposable subscribe() {
36273631
@CheckReturnValue
36283632
@SchedulerSupport(SchedulerSupport.NONE)
36293633
public final Disposable subscribe(Consumer<? super T> onSuccess) {
3630-
return subscribe(onSuccess, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION);
3634+
return subscribe(onSuccess, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION);
36313635
}
36323636

36333637
/**

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

Lines changed: 19 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -4917,6 +4917,10 @@ public final void blockingSubscribe() {
49174917

49184918
/**
49194919
* Subscribes to the source and calls the given callbacks <strong>on the current thread</strong>.
4920+
* <p>
4921+
* If the Observable emits an error, it is wrapped into an
4922+
* {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
4923+
* and routed to the RxJavaPlugins.onError handler.
49204924
* <dl>
49214925
* <dt><b>Scheduler:</b></dt>
49224926
* <dd>{@code blockingSubscribe} does not operate by default on a particular {@link Scheduler}.</dd>
@@ -4926,7 +4930,7 @@ public final void blockingSubscribe() {
49264930
*/
49274931
@SchedulerSupport(SchedulerSupport.NONE)
49284932
public final void blockingSubscribe(Consumer<? super T> onNext) {
4929-
ObservableBlockingSubscribe.subscribe(this, onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION);
4933+
ObservableBlockingSubscribe.subscribe(this, onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION);
49304934
}
49314935

49324936
/**
@@ -6761,7 +6765,7 @@ public final Observable<T> doFinally(Action onFinally) {
67616765
* The action is shared between subscriptions and thus may be called concurrently from multiple
67626766
* threads; the action must be thread safe.
67636767
* <p>
6764-
* If the action throws a runtime exception, that exception is rethrown by the {@code unsubscribe()} call,
6768+
* If the action throws a runtime exception, that exception is rethrown by the {@code dispose()} call,
67656769
* sometimes as a {@code CompositeException} if there were multiple exceptions along the way.
67666770
* <p>
67676771
* Note that terminal events trigger the action unless the {@code ObservableSource} is subscribed to via {@code unsafeSubscribe()}.
@@ -7785,6 +7789,10 @@ public final Disposable forEach(Consumer<? super T> onNext) {
77857789
/**
77867790
* Subscribes to the {@link ObservableSource} and receives notifications for each element until the
77877791
* onNext Predicate returns false.
7792+
* <p>
7793+
* If the Observable emits an error, it is wrapped into an
7794+
* {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
7795+
* and routed to the RxJavaPlugins.onError handler.
77887796
* <dl>
77897797
* <dt><b>Scheduler:</b></dt>
77907798
* <dd>{@code forEachWhile} does not operate by default on a particular {@link Scheduler}.</dd>
@@ -7801,7 +7809,7 @@ public final Disposable forEach(Consumer<? super T> onNext) {
78017809
@CheckReturnValue
78027810
@SchedulerSupport(SchedulerSupport.NONE)
78037811
public final Disposable forEachWhile(Predicate<? super T> onNext) {
7804-
return forEachWhile(onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION);
7812+
return forEachWhile(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION);
78057813
}
78067814

78077815
/**
@@ -10536,7 +10544,9 @@ public final Observable<T> startWithArray(T... items) {
1053610544
/**
1053710545
* Subscribes to an ObservableSource and ignores {@code onNext} and {@code onComplete} emissions.
1053810546
* <p>
10539-
* If the Observable emits an error, it is routed to the RxJavaPlugins.onError handler.
10547+
* If the Observable emits an error, it is wrapped into an
10548+
* {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
10549+
* and routed to the RxJavaPlugins.onError handler.
1054010550
* <dl>
1054110551
* <dt><b>Scheduler:</b></dt>
1054210552
* <dd>{@code subscribe} does not operate by default on a particular {@link Scheduler}.</dd>
@@ -10548,13 +10558,15 @@ public final Observable<T> startWithArray(T... items) {
1054810558
*/
1054910559
@SchedulerSupport(SchedulerSupport.NONE)
1055010560
public final Disposable subscribe() {
10551-
return subscribe(Functions.emptyConsumer(), Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION, Functions.emptyConsumer());
10561+
return subscribe(Functions.emptyConsumer(), Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());
1055210562
}
1055310563

1055410564
/**
1055510565
* Subscribes to an ObservableSource and provides a callback to handle the items it emits.
1055610566
* <p>
10557-
* If the Observable emits an error, it is routed to the RxJavaPlugins.onError handler.
10567+
* If the Observable emits an error, it is wrapped into an
10568+
* {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
10569+
* and routed to the RxJavaPlugins.onError handler.
1055810570
* <dl>
1055910571
* <dt><b>Scheduler:</b></dt>
1056010572
* <dd>{@code subscribe} does not operate by default on a particular {@link Scheduler}.</dd>
@@ -10571,7 +10583,7 @@ public final Disposable subscribe() {
1057110583
@CheckReturnValue
1057210584
@SchedulerSupport(SchedulerSupport.NONE)
1057310585
public final Disposable subscribe(Consumer<? super T> onNext) {
10574-
return subscribe(onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION, Functions.emptyConsumer());
10586+
return subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());
1057510587
}
1057610588

1057710589
/**

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

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2553,6 +2553,10 @@ public final Single<T> retryWhen(Function<? super Flowable<Throwable>, ? extends
25532553

25542554
/**
25552555
* Subscribes to a Single but ignore its emission or notification.
2556+
* <p>
2557+
* If the Single emits an error, it is wrapped into an
2558+
* {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
2559+
* and routed to the RxJavaPlugins.onError handler.
25562560
* <dl>
25572561
* <dt><b>Scheduler:</b></dt>
25582562
* <dd>{@code subscribe} does not operate by default on a particular {@link Scheduler}.</dd>
@@ -2563,7 +2567,7 @@ public final Single<T> retryWhen(Function<? super Flowable<Throwable>, ? extends
25632567
*/
25642568
@SchedulerSupport(SchedulerSupport.NONE)
25652569
public final Disposable subscribe() {
2566-
return subscribe(Functions.emptyConsumer(), Functions.ERROR_CONSUMER);
2570+
return subscribe(Functions.emptyConsumer(), Functions.ON_ERROR_MISSING);
25672571
}
25682572

25692573
/**
@@ -2594,6 +2598,10 @@ public final Disposable subscribe(final BiConsumer<? super T, ? super Throwable>
25942598

25952599
/**
25962600
* Subscribes to a Single and provides a callback to handle the item it emits.
2601+
* <p>
2602+
* If the Single emits an error, it is wrapped into an
2603+
* {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
2604+
* and routed to the RxJavaPlugins.onError handler.
25972605
* <dl>
25982606
* <dt><b>Scheduler:</b></dt>
25992607
* <dd>{@code subscribe} does not operate by default on a particular {@link Scheduler}.</dd>
@@ -2609,7 +2617,7 @@ public final Disposable subscribe(final BiConsumer<? super T, ? super Throwable>
26092617
@CheckReturnValue
26102618
@SchedulerSupport(SchedulerSupport.NONE)
26112619
public final Disposable subscribe(Consumer<? super T> onSuccess) {
2612-
return subscribe(onSuccess, Functions.ERROR_CONSUMER);
2620+
return subscribe(onSuccess, Functions.ON_ERROR_MISSING);
26132621
}
26142622

26152623
/**

‎src/main/java/io/reactivex/exceptions/OnErrorNotImplementedException.java

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
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.exceptions;
15+
16+
import io.reactivex.annotations.Experimental;
17+
18+
/**
19+
* Represents an exception used to signal to the {@code RxJavaPlugins.onError()} that a
20+
* callback-based subscribe() method on a base reactive type didn't specify
21+
* an onError handler.
22+
* @since 2.0.6 - experimental
23+
*/
24+
@Experimental
25+
public final class OnErrorNotImplementedException extends RuntimeException {
26+
27+
private static final long serialVersionUID = -6298857009889503852L;
28+
29+
/**
30+
* Customizes the {@code Throwable} with a custom message and wraps it before it
31+
* is signalled to the {@code RxJavaPlugins.onError()} handler as {@code OnErrorNotImplementedException}.
32+
*
33+
* @param message
34+
* the message to assign to the {@code Throwable} to signal
35+
* @param e
36+
* the {@code Throwable} to signal; if null, a NullPointerException is constructed
37+
*/
38+
public OnErrorNotImplementedException(String message, Throwable e) {
39+
super(message, e != null ? e : new NullPointerException());
40+
}
41+
42+
/**
43+
* Wraps the {@code Throwable} before it
44+
* is signalled to the {@code RxJavaPlugins.onError()}
45+
* handler as {@code OnErrorNotImplementedException}.
46+
*
47+
* @param e
48+
* the {@code Throwable} to signal; if null, a NullPointerException is constructed
49+
*/
50+
public OnErrorNotImplementedException(Throwable e) {
51+
super(e != null ? e.getMessage() : null, e != null ? e : new NullPointerException());
52+
}
53+
}

‎src/main/java/io/reactivex/internal/functions/Functions.java

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import org.reactivestreams.Subscription;
1919

2020
import io.reactivex.*;
21+
import io.reactivex.exceptions.OnErrorNotImplementedException;
2122
import io.reactivex.functions.*;
2223
import io.reactivex.plugins.RxJavaPlugins;
2324
import io.reactivex.schedulers.Timed;
@@ -218,6 +219,17 @@ public void accept(Throwable error) {
218219
}
219220
};
220221

222+
/**
223+
* Wraps the consumed Throwable into an OnErrorNotImplementedException and
224+
* signals it to the plugin error handler.
225+
*/
226+
public static final Consumer<Throwable> ON_ERROR_MISSING = new Consumer<Throwable>() {
227+
@Override
228+
public void accept(Throwable error) {
229+
RxJavaPlugins.onError(new OnErrorNotImplementedException(error));
230+
}
231+
};
232+
221233
public static final LongConsumer EMPTY_LONG_CONSUMER = new LongConsumer() {
222234
@Override
223235
public void accept(long v) { }

‎src/main/java/io/reactivex/internal/observers/CallbackCompletableObserver.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,9 @@
1717

1818
import io.reactivex.CompletableObserver;
1919
import io.reactivex.disposables.Disposable;
20-
import io.reactivex.exceptions.Exceptions;
20+
import io.reactivex.exceptions.*;
2121
import io.reactivex.functions.*;
22-
import io.reactivex.internal.disposables.*;
22+
import io.reactivex.internal.disposables.DisposableHelper;
2323
import io.reactivex.plugins.RxJavaPlugins;
2424

2525
public final class CallbackCompletableObserver
@@ -43,7 +43,7 @@ public CallbackCompletableObserver(Consumer<? super Throwable> onError, Action o
4343

4444
@Override
4545
public void accept(Throwable e) {
46-
RxJavaPlugins.onError(e);
46+
RxJavaPlugins.onError(new OnErrorNotImplementedException(e));
4747
}
4848

4949
@Override

‎src/main/java/io/reactivex/internal/observers/EmptyCompletableObserver.java

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,8 @@
1717

1818
import io.reactivex.CompletableObserver;
1919
import io.reactivex.disposables.Disposable;
20-
import io.reactivex.internal.disposables.*;
20+
import io.reactivex.exceptions.OnErrorNotImplementedException;
21+
import io.reactivex.internal.disposables.DisposableHelper;
2122
import io.reactivex.plugins.RxJavaPlugins;
2223

2324
public final class EmptyCompletableObserver
@@ -46,7 +47,7 @@ public void onComplete() {
4647
@Override
4748
public void onError(Throwable e) {
4849
lazySet(DisposableHelper.DISPOSED);
49-
RxJavaPlugins.onError(e);
50+
RxJavaPlugins.onError(new OnErrorNotImplementedException(e));
5051
}
5152

5253
@Override

0 commit comments

Comments
 (0)