ReactiveX · akarnokd · Dec 26, 2019 · Dec 26, 2019
diff --git a/src/main/java/io/reactivex/rxjava3/disposables/AutoCloseableDisposable.java b/src/main/java/io/reactivex/rxjava3/disposables/AutoCloseableDisposable.java
@@ -14,7 +14,7 @@
 package io.reactivex.rxjava3.disposables;
 
 import io.reactivex.rxjava3.annotations.NonNull;
-import io.reactivex.rxjava3.plugins.RxJavaPlugins;
+import io.reactivex.rxjava3.internal.util.ExceptionHelper;
 
 /**
  * A disposable container that manages an {@link AutoCloseable} instance.
@@ -33,7 +33,7 @@ protected void onDisposed(@NonNull AutoCloseable value) {
         try {
             value.close();
         } catch (Throwable ex) {
-            RxJavaPlugins.onError(ex);
+            throw ExceptionHelper.wrapOrThrow(ex);
         }
     }
 

diff --git a/src/main/java/io/reactivex/rxjava3/internal/jdk8/FlowableFlatMapStream.java b/src/main/java/io/reactivex/rxjava3/internal/jdk8/FlowableFlatMapStream.java
@@ -21,7 +21,7 @@
 
 import io.reactivex.rxjava3.annotations.NonNull;
 import io.reactivex.rxjava3.core.*;
-import io.reactivex.rxjava3.exceptions.MissingBackpressureException;
+import io.reactivex.rxjava3.exceptions.*;
 import io.reactivex.rxjava3.functions.*;
 import io.reactivex.rxjava3.internal.fuseable.*;
 import io.reactivex.rxjava3.internal.queue.SpscArrayQueue;
@@ -61,6 +61,7 @@ protected void subscribeActual(Subscriber<? super R> s) {
                     stream = Objects.requireNonNull(mapper.apply(t), "The mapper returned a null Stream");
                 }
             } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
                 EmptySubscription.error(ex, s);
                 return;
             }
@@ -243,6 +244,7 @@ void drain() {
                         try {
                             t = queue.poll();
                         } catch (Throwable ex) {
+                            Exceptions.throwIfFatal(ex);
                             trySignalError(downstream, ex);
                             continue;
                         }
@@ -271,6 +273,7 @@ else if (!isEmpty) {
                                     iterator = null;
                                 }
                             } catch (Throwable ex) {
+                                Exceptions.throwIfFatal(ex);
                                 trySignalError(downstream, ex);
                             }
                             continue;
@@ -282,6 +285,7 @@ else if (!isEmpty) {
                         try {
                             item = Objects.requireNonNull(iterator.next(), "The Stream.Iterator returned a null value");
                         } catch (Throwable ex) {
+                            Exceptions.throwIfFatal(ex);
                             trySignalError(downstream, ex);
                             continue;
                         }
@@ -297,6 +301,7 @@ else if (!isEmpty) {
                                         clearCurrentRethrowCloseError();
                                     }
                                 } catch (Throwable ex) {
+                                    Exceptions.throwIfFatal(ex);
                                     trySignalError(downstream, ex);
                                 }
                             }
@@ -328,6 +333,7 @@ void clearCurrentSuppressCloseError() {
             try {
                 clearCurrentRethrowCloseError();
             } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
                 RxJavaPlugins.onError(ex);
             }
         }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/jdk8/ObservableFlatMapStream.java b/src/main/java/io/reactivex/rxjava3/internal/jdk8/ObservableFlatMapStream.java
@@ -55,6 +55,7 @@ protected void subscribeActual(Observer<? super R> observer) {
                     stream = Objects.requireNonNull(mapper.apply(t), "The mapper returned a null Stream");
                 }
             } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
                 EmptyDisposable.error(ex, observer);
                 return;
             }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableCollect.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableCollect.java
@@ -39,6 +39,7 @@ protected void subscribeActual(Subscriber<? super U> s) {
         try {
             u = Objects.requireNonNull(initialSupplier.get(), "The initial value supplied is null");
         } catch (Throwable e) {
+            Exceptions.throwIfFatal(e);
             EmptySubscription.error(e, s);
             return;
         }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableCollectSingle.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableCollectSingle.java
@@ -44,6 +44,7 @@ protected void subscribeActual(SingleObserver<? super U> observer) {
         try {
             u = Objects.requireNonNull(initialSupplier.get(), "The initialSupplier returned a null value");
         } catch (Throwable e) {
+            Exceptions.throwIfFatal(e);
             EmptyDisposable.error(e, observer);
             return;
         }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDoOnEach.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDoOnEach.java
@@ -159,6 +159,7 @@ public T poll() throws Throwable {
                 try {
                     onError.accept(ex);
                 } catch (Throwable exc) {
+                    Exceptions.throwIfFatal(exc);
                     throw new CompositeException(ex, exc);
                 }
                 throw ExceptionHelper.<Exception>throwIfThrowable(ex);
@@ -173,6 +174,7 @@ public T poll() throws Throwable {
                         try {
                             onError.accept(ex);
                         } catch (Throwable exc) {
+                            Exceptions.throwIfFatal(exc);
                             throw new CompositeException(ex, exc);
                         }
                         throw ExceptionHelper.<Exception>throwIfThrowable(ex);
@@ -314,6 +316,7 @@ public T poll() throws Throwable {
                 try {
                     onError.accept(ex);
                 } catch (Throwable exc) {
+                    Exceptions.throwIfFatal(exc);
                     throw new CompositeException(ex, exc);
                 }
                 throw ExceptionHelper.<Exception>throwIfThrowable(ex);
@@ -328,6 +331,7 @@ public T poll() throws Throwable {
                         try {
                             onError.accept(ex);
                         } catch (Throwable exc) {
+                            Exceptions.throwIfFatal(exc);
                             throw new CompositeException(ex, exc);
                         }
                         throw ExceptionHelper.<Exception>throwIfThrowable(ex);

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableReplay.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableReplay.java
@@ -209,6 +209,7 @@ public void connect(Consumer<? super Disposable> connection) {
         try {
             connection.accept(ps);
         } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
             if (doConnect) {
                 ps.shouldConnect.compareAndSet(true, false);
             }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableScalarXMap.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableScalarXMap.java
@@ -136,6 +136,7 @@ public void subscribeActual(Subscriber<? super R> s) {
             try {
                 other = Objects.requireNonNull(mapper.apply(value), "The mapper returned a null Publisher");
             } catch (Throwable e) {
+                Exceptions.throwIfFatal(e);
                 EmptySubscription.error(e, s);
                 return;
             }

diff --git a/...java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowBoundarySelector.java b/...java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowBoundarySelector.java
@@ -245,6 +245,7 @@ void drain() {
                                     try {
                                         endSource = Objects.requireNonNull(closingIndicator.apply(startItem), "The closingIndicator returned a null Publisher");
                                     } catch (Throwable ex) {
+                                        Exceptions.throwIfFatal(ex);
                                         upstream.cancel();
                                         startSubscriber.cancel();
                                         resources.dispose();

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableZipIterable.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableZipIterable.java
@@ -101,15 +101,15 @@ public void onNext(T t) {
             try {
                 u = Objects.requireNonNull(iterator.next(), "The iterator returned a null value");
             } catch (Throwable e) {
-                error(e);
+                fail(e);
                 return;
             }
 
             V v;
             try {
                 v = Objects.requireNonNull(zipper.apply(t, u), "The zipper function returned a null value");
             } catch (Throwable e) {
-                error(e);
+                fail(e);
                 return;
             }
 
@@ -120,7 +120,7 @@ public void onNext(T t) {
             try {
                 b = iterator.hasNext();
             } catch (Throwable e) {
-                error(e);
+                fail(e);
                 return;
             }
 
@@ -131,7 +131,7 @@ public void onNext(T t) {
             }
         }
 
-        void error(Throwable e) {
+        void fail(Throwable e) {
             Exceptions.throwIfFatal(e);
             done = true;
             upstream.cancel();

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeFromFuture.java b/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeFromFuture.java
@@ -52,6 +52,7 @@ protected void subscribeActual(MaybeObserver<? super T> observer) {
                     v = future.get(timeout, unit);
                 }
             } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
                 if (ex instanceof ExecutionException) {
                     ex = ex.getCause();
                 }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableBuffer.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableBuffer.java
@@ -186,6 +186,7 @@ public void onNext(T t) {
                 try {
                     b = ExceptionHelper.nullCheck(bufferSupplier.get(), "The bufferSupplier returned a null Collection.");
                 } catch (Throwable e) {
+                    Exceptions.throwIfFatal(e);
                     buffers.clear();
                     upstream.dispose();
                     downstream.onError(e);

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCollect.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCollect.java
@@ -14,6 +14,7 @@
 
 import io.reactivex.rxjava3.core.*;
 import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.exceptions.Exceptions;
 import io.reactivex.rxjava3.functions.*;
 import io.reactivex.rxjava3.internal.disposables.*;
 import io.reactivex.rxjava3.plugins.RxJavaPlugins;
@@ -37,6 +38,7 @@ protected void subscribeActual(Observer<? super U> t) {
         try {
             u = Objects.requireNonNull(initialSupplier.get(), "The initialSupplier returned a null value");
         } catch (Throwable e) {
+            Exceptions.throwIfFatal(e);
             EmptyDisposable.error(e, t);
             return;
         }
@@ -86,6 +88,7 @@ public void onNext(T t) {
             try {
                 collector.accept(u, t);
             } catch (Throwable e) {
+                Exceptions.throwIfFatal(e);
                 upstream.dispose();
                 onError(e);
             }

diff --git a/...main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCollectSingle.java b/...main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCollectSingle.java
@@ -14,6 +14,7 @@
 
 import io.reactivex.rxjava3.core.*;
 import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.exceptions.Exceptions;
 import io.reactivex.rxjava3.functions.*;
 import io.reactivex.rxjava3.internal.disposables.*;
 import io.reactivex.rxjava3.internal.fuseable.FuseToObservable;
@@ -41,6 +42,7 @@ protected void subscribeActual(SingleObserver<? super U> t) {
         try {
             u = Objects.requireNonNull(initialSupplier.get(), "The initialSupplier returned a null value");
         } catch (Throwable e) {
+            Exceptions.throwIfFatal(e);
             EmptyDisposable.error(e, t);
             return;
         }
@@ -94,6 +96,7 @@ public void onNext(T t) {
             try {
                 collector.accept(u, t);
             } catch (Throwable e) {
+                Exceptions.throwIfFatal(e);
                 upstream.dispose();
                 onError(e);
             }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableReplay.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableReplay.java
@@ -205,6 +205,7 @@ public void connect(Consumer<? super Disposable> connection) {
         try {
             connection.accept(ps);
         } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
             if (doConnect) {
                 ps.shouldConnect.compareAndSet(true, false);
             }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableScalarXMap.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableScalarXMap.java
@@ -140,6 +140,7 @@ public void subscribeActual(Observer<? super R> observer) {
             try {
                 other = Objects.requireNonNull(mapper.apply(value), "The mapper returned a null ObservableSource");
             } catch (Throwable e) {
+                Exceptions.throwIfFatal(e);
                 EmptyDisposable.error(e, observer);
                 return;
             }

diff --git a/.../io/reactivex/rxjava3/internal/operators/observable/ObservableWindowBoundarySelector.java b/.../io/reactivex/rxjava3/internal/operators/observable/ObservableWindowBoundarySelector.java
@@ -237,6 +237,7 @@ void drain() {
                                 try {
                                     endSource = Objects.requireNonNull(closingIndicator.apply(startItem), "The closingIndicator returned a null ObservableSource");
                                 } catch (Throwable ex) {
+                                    Exceptions.throwIfFatal(ex);
                                     upstream.dispose();
                                     startObserver.dispose();
                                     resources.dispose();

diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/InstantPeriodicTask.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/InstantPeriodicTask.java
@@ -20,6 +20,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.exceptions.Exceptions;
 import io.reactivex.rxjava3.internal.functions.Functions;
 import io.reactivex.rxjava3.plugins.RxJavaPlugins;
 
@@ -56,6 +57,7 @@ public Void call() throws Exception {
             setRest(executor.submit(this));
             runner = null;
         } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
             runner = null;
             RxJavaPlugins.onError(ex);
         }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/ScheduledDirectPeriodicTask.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/ScheduledDirectPeriodicTask.java
@@ -16,6 +16,7 @@
 
 package io.reactivex.rxjava3.internal.schedulers;
 
+import io.reactivex.rxjava3.exceptions.Exceptions;
 import io.reactivex.rxjava3.plugins.RxJavaPlugins;
 
 /**
@@ -38,6 +39,7 @@ public void run() {
             runnable.run();
             runner = null;
         } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
             runner = null;
             lazySet(FINISHED);
             RxJavaPlugins.onError(ex);

diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/SchedulerPoolFactory.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/SchedulerPoolFactory.java
@@ -20,6 +20,7 @@
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicReference;
 
+import io.reactivex.rxjava3.exceptions.Exceptions;
 import io.reactivex.rxjava3.functions.Function;
 
 /**
@@ -108,6 +109,7 @@ static int getIntProperty(boolean enabled, String key, int defaultNotFound, int
                 }
                 return Integer.parseInt(value);
             } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
                 return defaultNotFound;
             }
         }
@@ -123,6 +125,7 @@ static boolean getBooleanProperty(boolean enabled, String key, boolean defaultNo
                 }
                 return "true".equals(value);
             } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
                 return defaultNotFound;
             }
         }

diff --git a/src/test/java/io/reactivex/rxjava3/disposables/DisposableTest.java b/src/test/java/io/reactivex/rxjava3/disposables/DisposableTest.java
@@ -213,17 +213,20 @@ public void fromAutoCloseableThrows() throws Throwable {
 
             assertTrue(errors.isEmpty());
 
-            d.dispose();
+            try {
+                d.dispose();
+                fail("Should have thrown!");
+            } catch (TestException expected) {
+                // expected
+            }
 
             assertTrue(d.isDisposed());
-            assertEquals(1, errors.size());
 
             d.dispose();
 
             assertTrue(d.isDisposed());
-            assertEquals(1, errors.size());
 
-            TestHelper.assertUndeliverable(errors, 0, TestException.class);
+            assertTrue(errors.isEmpty());
         });
     }