2.x: Functional interfaces now throw (#4278)

* 1.x: use throwing functional interfaces + changes

* Fix remaining classes

* Fix mistakes.

Author: akarnokd

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

@@ -16,12 +16,13 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.annotations.*;
+import io.reactivex.annotations.SchedulerSupport;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.functions.*;
 import io.reactivex.internal.functions.*;
 import io.reactivex.internal.operators.completable.*;
 import io.reactivex.internal.subscribers.completable.*;
+import io.reactivex.internal.util.Exceptions;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.schedulers.Schedulers;
 
@@ -200,7 +201,7 @@ public static Completable create(CompletableConsumable onSubscribe) {
      * @return the Completable instance
      */
     @SchedulerSupport(SchedulerSupport.NONE)
-    public static Completable defer(final Supplier<? extends CompletableConsumable> completableSupplier) {
+    public static Completable defer(final Callable<? extends CompletableConsumable> completableSupplier) {
         Objects.requireNonNull(completableSupplier, "completableSupplier");
         return new CompletableDefer(completableSupplier);
     }
@@ -216,7 +217,7 @@ public static Completable defer(final Supplier<? extends CompletableConsumable>
      * @throws NullPointerException if errorSupplier is null
      */
     @SchedulerSupport(SchedulerSupport.NONE)
-    public static Completable error(final Supplier<? extends Throwable> errorSupplier) {
+    public static Completable error(final Callable<? extends Throwable> errorSupplier) {
         Objects.requireNonNull(errorSupplier, "errorSupplier is null");
         return new CompletableErrorSupplier(errorSupplier);
     }
@@ -499,7 +500,7 @@ private static NullPointerException toNpe(Throwable ex) {
      * @param disposer the consumer that disposes the resource created by the resource supplier
      * @return the new Completable instance
      */
-    public static <R> Completable using(Supplier<R> resourceSupplier, 
+    public static <R> Completable using(Callable<R> resourceSupplier, 
             Function<? super R, ? extends CompletableConsumable> completableFunction, 
             Consumer<? super R> disposer) {
         return using(resourceSupplier, completableFunction, disposer, true);
@@ -523,7 +524,7 @@ public static <R> Completable using(Supplier<R> resourceSupplier,
      * @return the new Completable instance
      */
     public static <R> Completable using(
-            final Supplier<R> resourceSupplier, 
+            final Callable<R> resourceSupplier, 
             final Function<? super R, ? extends CompletableConsumable> completableFunction, 
             final Consumer<? super R> disposer, 
             final boolean eager) {
@@ -1315,7 +1316,11 @@ private Completable timeout0(long timeout, TimeUnit unit, Scheduler scheduler, C
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <U> U to(Function<? super Completable, U> converter) {
-        return converter.apply(this);
+        try {
+            return converter.apply(this);
+        } catch (Throwable ex) {
+            throw Exceptions.propagate(ex);
+        }
     }
 
     /**
@@ -1349,7 +1354,7 @@ public final <T> Observable<T> toObservable() {
      * @throws NullPointerException if completionValueSupplier is null
      */
     @SchedulerSupport(SchedulerSupport.NONE)
-    public final <T> Single<T> toSingle(final Supplier<? extends T> completionValueSupplier) {
+    public final <T> Single<T> toSingle(final Callable<? extends T> completionValueSupplier) {
         Objects.requireNonNull(completionValueSupplier, "completionValueSupplier is null");
         return new CompletableToSingle<T>(this, completionValueSupplier, null);
     }

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

@@ -24,6 +24,7 @@
 import io.reactivex.internal.functions.Objects;
 import io.reactivex.internal.operators.single.*;
 import io.reactivex.internal.subscribers.single.*;
+import io.reactivex.internal.util.Exceptions;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.schedulers.Schedulers;
 
@@ -60,9 +61,9 @@ public static <T> Single<T> amb(final Iterable<? extends SingleConsumable<? exte
     @SuppressWarnings("unchecked")
     public static <T> Single<T> amb(final SingleConsumable<? extends T>... sources) {
         if (sources.length == 0) {
-            return error(new Supplier<Throwable>() {
+            return error(new Callable<Throwable>() {
                 @Override
-                public Throwable get() {
+                public Throwable call() {
                     return new NoSuchElementException();
                 }
             });
@@ -208,21 +209,21 @@ public static <T> Single<T> create(SingleConsumable<T> onSubscribe) {
         return new SingleWrapper<T>(onSubscribe);
     }
 
-    public static <T> Single<T> defer(final Supplier<? extends SingleConsumable<? extends T>> singleSupplier) {
+    public static <T> Single<T> defer(final Callable<? extends SingleConsumable<? extends T>> singleSupplier) {
         Objects.requireNonNull(singleSupplier, "singleSupplier is null");
         return new SingleDefer<T>(singleSupplier);
     }
 
-    public static <T> Single<T> error(final Supplier<? extends Throwable> errorSupplier) {
+    public static <T> Single<T> error(final Callable<? extends Throwable> errorSupplier) {
         Objects.requireNonNull(errorSupplier, "errorSupplier is null");
         return new SingleError<T>(errorSupplier);
     }
 
     public static <T> Single<T> error(final Throwable error) {
         Objects.requireNonNull(error, "error is null");
-        return error(new Supplier<Throwable>() {
+        return error(new Callable<Throwable>() {
             @Override
-            public Throwable get() {
+            public Throwable call() {
                 return error;
             }
         });
@@ -415,13 +416,13 @@ public static <T> Single<Boolean> equals(final SingleConsumable<? extends T> fir
         return new SingleEquals<T>(first, second);
     }
 
-    public static <T, U> Single<T> using(Supplier<U> resourceSupplier, 
+    public static <T, U> Single<T> using(Callable<U> resourceSupplier, 
             Function<? super U, ? extends SingleConsumable<? extends T>> singleFunction, Consumer<? super U> disposer) {
         return using(resourceSupplier, singleFunction, disposer, true);
     }
 
     public static <T, U> Single<T> using(
-            final Supplier<U> resourceSupplier, 
+            final Callable<U> resourceSupplier, 
             final Function<? super U, ? extends SingleConsumable<? extends T>> singleFunction, 
             final Consumer<? super U> disposer, 
             final boolean eager) {
@@ -730,7 +731,7 @@ public final Single<T> observeOn(final Scheduler scheduler) {
         return new SingleObserveOn<T>(this, scheduler);
     }
 
-    public final Single<T> onErrorReturn(final Supplier<? extends T> valueSupplier) {
+    public final Single<T> onErrorReturn(final Callable<? extends T> valueSupplier) {
         Objects.requireNonNull(valueSupplier, "valueSupplier is null");
         return new SingleOnErrorReturn<T>(this, valueSupplier, null);
     }
@@ -858,7 +859,11 @@ private Single<T> timeout0(final long timeout, final TimeUnit unit, final Schedu
     }
 
     public final <R> R to(Function<? super Single<T>, R> convert) {
-        return convert.apply(this);
+        try {
+            return convert.apply(this);
+        } catch (Throwable ex) {
+            throw Exceptions.propagate(ex);
+        }
     }
 
     public final Flowable<T> toFlowable() {

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

@@ -373,19 +373,48 @@ public void subscribe(final Consumer<? super T> onNext, final Consumer<? super T
      */
     public void subscribe(final Consumer<? super T> onNext, final Consumer<? super Throwable> onError, final Runnable onComplete) {
         subscribe(new DefaultObserver<T>() {
+            boolean done;
             @Override
             public void onNext(T t) {
-                onNext.accept(t);
+                if (done) {
+                    return;
+                }
+                try {
+                    onNext.accept(t);
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    cancel();
+                    onError(ex);
+                }
             }
 
             @Override
             public void onError(Throwable e) {
-                onError.accept(e);
+                if (done) {
+                    RxJavaPlugins.onError(e);
+                    return;
+                }
+                done = true;
+                try {
+                    onError.accept(e);
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    RxJavaPlugins.onError(ex);
+                }
             }
 
             @Override
             public void onComplete() {
-                onComplete.run();
+                if (done) {
+                    return;
+                }
+                done = true;
+                try {
+                    onComplete.run();
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    RxJavaPlugins.onError(ex);
+                }
             }
         });
     }

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

@@ -15,5 +15,5 @@
 
 public interface BiConsumer<T1, T2> {
 
-    void accept(T1 t1, T2 t2);
+    void accept(T1 t1, T2 t2) throws Exception;
 }

src/main/java/io/reactivex/flowables/BlockingFlowable.java

@@ -15,5 +15,5 @@
 
 public interface BiFunction<T1, T2, R> {
 
-    R apply(T1 t1, T2 t2);
+    R apply(T1 t1, T2 t2) throws Exception;
 }

src/main/java/io/reactivex/functions/BiConsumer.java

@@ -15,5 +15,5 @@
 
 public interface BiPredicate<T1, T2> {
 
-    boolean test(T1 t1, T2 t2);
+    boolean test(T1 t1, T2 t2) throws Exception;
 }

src/main/java/io/reactivex/functions/BiFunction.java

@@ -14,5 +14,5 @@
 package io.reactivex.functions;
 
 public interface BooleanSupplier {
-    boolean getAsBoolean(); // NOPMD
+    boolean getAsBoolean() throws Exception; // NOPMD
 }

src/main/java/io/reactivex/functions/BiPredicate.java

@@ -14,5 +14,5 @@
 package io.reactivex.functions;
 
 public interface Consumer<T> {
-    void accept(T t);
+    void accept(T t) throws Exception;
 }

src/main/java/io/reactivex/functions/BooleanSupplier.java

@@ -14,5 +14,5 @@
 package io.reactivex.functions;
 
 public interface Function<T, R> {
-    R apply(T t);
+    R apply(T t) throws Exception;
 }

src/main/java/io/reactivex/functions/Consumer.java

@@ -14,5 +14,5 @@
 package io.reactivex.functions;
 
 public interface Function3<T1, T2, T3, R> {
-    R apply(T1 t1, T2 t2, T3 t3);
+    R apply(T1 t1, T2 t2, T3 t3) throws Exception;
 }

src/main/java/io/reactivex/functions/Function.java

@@ -14,5 +14,5 @@
 package io.reactivex.functions;
 
 public interface Function4<T1, T2, T3, T4, R> {
-    R apply(T1 t1, T2 t2, T3 t3, T4 t4);
+    R apply(T1 t1, T2 t2, T3 t3, T4 t4) throws Exception;
 }

src/main/java/io/reactivex/functions/Function3.java

@@ -14,5 +14,5 @@
 package io.reactivex.functions;
 
 public interface Function5<T1, T2, T3, T4, T5, R> {
-    R apply(T1 t1, T2 t2, T3 t3, T4 t4, T5 t5);
+    R apply(T1 t1, T2 t2, T3 t3, T4 t4, T5 t5) throws Exception;
 }

src/main/java/io/reactivex/functions/Function4.java

@@ -14,5 +14,5 @@
 package io.reactivex.functions;
 
 public interface Function6<T1, T2, T3, T4, T5, T6, R> {
-    R apply(T1 t1, T2 t2, T3 t3, T4 t4, T5 t5, T6 t6);
+    R apply(T1 t1, T2 t2, T3 t3, T4 t4, T5 t5, T6 t6) throws Exception;
 }

src/main/java/io/reactivex/functions/Function5.java

@@ -14,5 +14,5 @@
 package io.reactivex.functions;
 
 public interface Function7<T1, T2, T3, T4, T5, T6, T7, R> {
-    R apply(T1 t1, T2 t2, T3 t3, T4 t4, T5 t5, T6 t6, T7 t7);
+    R apply(T1 t1, T2 t2, T3 t3, T4 t4, T5 t5, T6 t6, T7 t7) throws Exception;
 }

src/main/java/io/reactivex/functions/Function6.java

@@ -14,5 +14,5 @@
 package io.reactivex.functions;
 
 public interface Function8<T1, T2, T3, T4, T5, T6, T7, T8, R> {
-    R apply(T1 t1, T2 t2, T3 t3, T4 t4, T5 t5, T6 t6, T7 t7, T8 t8);
+    R apply(T1 t1, T2 t2, T3 t3, T4 t4, T5 t5, T6 t6, T7 t7, T8 t8) throws Exception;
 }

src/main/java/io/reactivex/functions/Function7.java

@@ -14,5 +14,5 @@
 package io.reactivex.functions;
 
 public interface Function9<T1, T2, T3, T4, T5, T6, T7, T8, T9, R> {
-    R apply(T1 t1, T2 t2, T3 t3, T4 t4, T5 t5, T6 t6, T7 t7, T8 t8, T9 t9);
+    R apply(T1 t1, T2 t2, T3 t3, T4 t4, T5 t5, T6 t6, T7 t7, T8 t8, T9 t9) throws Exception;
 }

src/main/java/io/reactivex/functions/Function8.java

@@ -13,5 +13,5 @@
 package io.reactivex.functions;
 
 public interface IntFunction<T> {
-    T apply(int i);
+    T apply(int i) throws Exception;
 }

src/main/java/io/reactivex/functions/Function9.java

@@ -13,5 +13,5 @@
 package io.reactivex.functions;
 
 public interface LongConsumer {
-    void accept(long t);
+    void accept(long t) throws Exception;
 }

src/main/java/io/reactivex/functions/IntFunction.java

@@ -14,5 +14,5 @@
 package io.reactivex.functions;
 
 public interface Predicate<T> {
-    boolean test(T t);
+    boolean test(T t) throws Exception;
 }

src/main/java/io/reactivex/functions/LongConsumer.java

@@ -112,13 +112,11 @@ void drain() {
         for (;;) {
 
             for (;;) {
-                Object o = q.peek();
-                
+                Object o = q.poll();
                 if (o == null) {
                     break;
                 }
 
-                q.poll();
                 Object v = q.poll();
 
                 if (o != s) {