Move error consumer helper to internal API. (#4386)

Author: JakeWharton

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

@@ -5410,7 +5410,7 @@ public final void blockingSubscribe() {
      * @since 2.0
      */
     public final void blockingSubscribe(Consumer<? super T> onNext) {
-        FlowableBlockingSubscribe.subscribe(this, onNext, RxJavaPlugins.errorConsumer(), Functions.EMPTY_ACTION);
+        FlowableBlockingSubscribe.subscribe(this, onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION);
     }
 
     /**
@@ -8496,7 +8496,7 @@ public final Disposable forEach(Consumer<? super T> onNext) {
     @BackpressureSupport(BackpressureKind.NONE)
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Disposable forEachWhile(Predicate<? super T> onNext) {
-        return forEachWhile(onNext, RxJavaPlugins.errorConsumer(), Functions.EMPTY_ACTION);
+        return forEachWhile(onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION);
     }
 
     /**
@@ -11837,7 +11837,7 @@ public final Flowable<T> startWithArray(T... values) {
     @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Disposable subscribe() {
-        return subscribe(Functions.emptyConsumer(), RxJavaPlugins.errorConsumer(), 
+        return subscribe(Functions.emptyConsumer(), Functions.ERROR_CONSUMER,
                 Functions.EMPTY_ACTION, FlowableInternalHelper.requestMax());
     }
 
@@ -11864,7 +11864,7 @@ public final Disposable subscribe() {
     @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Disposable subscribe(Consumer<? super T> onNext) {
-        return subscribe(onNext, RxJavaPlugins.errorConsumer(), 
+        return subscribe(onNext, Functions.ERROR_CONSUMER,
                 Functions.EMPTY_ACTION, FlowableInternalHelper.requestMax());
     }
 

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

@@ -4870,7 +4870,7 @@ public final void blockingSubscribe() {
      * @since 2.0
      */
     public final void blockingSubscribe(Consumer<? super T> onNext) {
-        ObservableBlockingSubscribe.subscribe(this, onNext, RxJavaPlugins.errorConsumer(), Functions.EMPTY_ACTION);
+        ObservableBlockingSubscribe.subscribe(this, onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION);
     }
 
     /**
@@ -7478,7 +7478,7 @@ public final Disposable forEach(Consumer<? super T> onNext) {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Disposable forEachWhile(Predicate<? super T> onNext) {
-        return forEachWhile(onNext, RxJavaPlugins.errorConsumer(), Functions.EMPTY_ACTION);
+        return forEachWhile(onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION);
     }
 
     /**
@@ -10043,7 +10043,7 @@ public final Observable<T> startWithArray(T... values) {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Disposable subscribe() {
-        return subscribe(Functions.emptyConsumer(), RxJavaPlugins.errorConsumer(), Functions.EMPTY_ACTION, Functions.emptyConsumer());
+        return subscribe(Functions.emptyConsumer(), Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION, Functions.emptyConsumer());
     }
 
     /**
@@ -10065,7 +10065,7 @@ public final Disposable subscribe() {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Disposable subscribe(Consumer<? super T> onNext) {
-        return subscribe(onNext, RxJavaPlugins.errorConsumer(), Functions.EMPTY_ACTION, Functions.emptyConsumer());
+        return subscribe(onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION, Functions.emptyConsumer());
     }
 
     /**

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

@@ -2511,7 +2511,7 @@ public final void safeSubscribe(Subscriber<? super T> s) {
      * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
      */
     public final Disposable subscribe() {
-        return subscribe(Functions.emptyConsumer(), RxJavaPlugins.errorConsumer());
+        return subscribe(Functions.emptyConsumer(), Functions.ERROR_CONSUMER);
     }
 
     /**
@@ -2553,7 +2553,7 @@ public final Disposable subscribe(final BiConsumer<? super T, ? super Throwable>
      * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
      */
     public final Disposable subscribe(Consumer<? super T> onSuccess) {
-        return subscribe(onSuccess, RxJavaPlugins.errorConsumer());
+        return subscribe(onSuccess, Functions.ERROR_CONSUMER);
     }
 
     /**

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

@@ -17,6 +17,7 @@
 
 import io.reactivex.*;
 import io.reactivex.functions.*;
+import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.schedulers.Timed;
 
 /**
@@ -183,7 +184,14 @@ public void accept(Object v) { }
     public static <T> Consumer<T> emptyConsumer() {
         return (Consumer<T>)EMPTY_CONSUMER;
     }
-    
+
+    public static final Consumer<Throwable> ERROR_CONSUMER = new Consumer<Throwable>() {
+        @Override
+        public void accept(Throwable error) {
+            RxJavaPlugins.onError(error);
+        }
+    };
+
     public static final LongConsumer EMPTY_LONGCONSUMER = new LongConsumer() {
         @Override
         public void accept(long v) { }

src/main/java/io/reactivex/plugins/RxJavaPlugins.java

@@ -747,22 +747,6 @@ public static Completable onAssembly(Completable source) {
         return source;
     }
 
-    /** Singleton consumer that calls RxJavaPlugins.onError. */
-    static final Consumer<Throwable> CONSUME_BY_RXJAVA_PLUGIN = new Consumer<Throwable>() {
-        @Override
-        public void accept(Throwable e) {
-            RxJavaPlugins.onError(e);
-        }
-    };
-    
-    /**
-     * Returns a consumer which relays the received Throwable to RxJavaPlugins.onError().
-     * @return the consumer
-     */
-    public static Consumer<Throwable> errorConsumer() {
-        return CONSUME_BY_RXJAVA_PLUGIN;
-    }
-    
     /**
      * Wraps the call to the function in try-catch and propagates thrown
      * checked exceptions as runtimeexception.