3.x: Add missing throwIfFatal calls (#6801)

akarnokd · web-flow · commit a97d871ee716 · 2019-12-27T00:55:36.000+01:00
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/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowBoundarySelector.java b/src/main/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/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCollectSingle.java b/src/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/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableWindowBoundarySelector.java b/src/main/java/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());
         });
     }
 
diff --git a/src/test/java/io/reactivex/rxjava3/validators/CatchThrowIfFatalCheck.java b/src/test/java/io/reactivex/rxjava3/validators/CatchThrowIfFatalCheck.java

Original file line number	Diff line number	Diff line change
`@@ -14,7 +14,7 @@`
`14`	`14`	`package io.reactivex.rxjava3.disposables;`
`15`	`15`
`16`	`16`	`import io.reactivex.rxjava3.annotations.NonNull;`
`17`		`-import io.reactivex.rxjava3.plugins.RxJavaPlugins;`
	`17`	`+import io.reactivex.rxjava3.internal.util.ExceptionHelper;`
`18`	`18`
`19`	`19`	`/**`
`20`	`20`	`* A disposable container that manages an {@link AutoCloseable} instance.`
`@@ -33,7 +33,7 @@ protected void onDisposed(@NonNull AutoCloseable value) {`
`33`	`33`	`try {`
`34`	`34`	`value.close();`
`35`	`35`	`} catch (Throwable ex) {`
`36`		`- RxJavaPlugins.onError(ex);`
	`36`	`+ throw ExceptionHelper.wrapOrThrow(ex);`
`37`	`37`	`}`
`38`	`38`	`}`
`39`	`39`
Original file line number	Diff line number	Diff line change
`@@ -21,7 +21,7 @@`
`21`	`21`
`22`	`22`	`import io.reactivex.rxjava3.annotations.NonNull;`
`23`	`23`	`import io.reactivex.rxjava3.core.*;`
`24`		`-import io.reactivex.rxjava3.exceptions.MissingBackpressureException;`
	`24`	`+import io.reactivex.rxjava3.exceptions.*;`
`25`	`25`	`import io.reactivex.rxjava3.functions.*;`
`26`	`26`	`import io.reactivex.rxjava3.internal.fuseable.*;`
`27`	`27`	`import io.reactivex.rxjava3.internal.queue.SpscArrayQueue;`
`@@ -61,6 +61,7 @@ protected void subscribeActual(Subscriber<? super R> s) {`
`61`	`61`	`stream = Objects.requireNonNull(mapper.apply(t), "The mapper returned a null Stream");`
`62`	`62`	`}`
`63`	`63`	`} catch (Throwable ex) {`
	`64`	`+ Exceptions.throwIfFatal(ex);`
`64`	`65`	`EmptySubscription.error(ex, s);`
`65`	`66`	`return;`
`66`	`67`	`}`
`@@ -243,6 +244,7 @@ void drain() {`
`243`	`244`	`try {`
`244`	`245`	`t = queue.poll();`
`245`	`246`	`} catch (Throwable ex) {`
	`247`	`+ Exceptions.throwIfFatal(ex);`
`246`	`248`	`trySignalError(downstream, ex);`
`247`	`249`	`continue;`
`248`	`250`	`}`
`@@ -271,6 +273,7 @@ else if (!isEmpty) {`
`271`	`273`	`iterator = null;`
`272`	`274`	`}`
`273`	`275`	`} catch (Throwable ex) {`
	`276`	`+ Exceptions.throwIfFatal(ex);`
`274`	`277`	`trySignalError(downstream, ex);`
`275`	`278`	`}`
`276`	`279`	`continue;`
`@@ -282,6 +285,7 @@ else if (!isEmpty) {`
`282`	`285`	`try {`
`283`	`286`	`item = Objects.requireNonNull(iterator.next(), "The Stream.Iterator returned a null value");`
`284`	`287`	`} catch (Throwable ex) {`
	`288`	`+ Exceptions.throwIfFatal(ex);`
`285`	`289`	`trySignalError(downstream, ex);`
`286`	`290`	`continue;`
`287`	`291`	`}`
`@@ -297,6 +301,7 @@ else if (!isEmpty) {`
`297`	`301`	`clearCurrentRethrowCloseError();`
`298`	`302`	`}`
`299`	`303`	`} catch (Throwable ex) {`
	`304`	`+ Exceptions.throwIfFatal(ex);`
`300`	`305`	`trySignalError(downstream, ex);`
`301`	`306`	`}`
`302`	`307`	`}`
`@@ -328,6 +333,7 @@ void clearCurrentSuppressCloseError() {`
`328`	`333`	`try {`
`329`	`334`	`clearCurrentRethrowCloseError();`
`330`	`335`	`} catch (Throwable ex) {`
	`336`	`+ Exceptions.throwIfFatal(ex);`
`331`	`337`	`RxJavaPlugins.onError(ex);`
`332`	`338`	`}`
`333`	`339`	`}`
Original file line number	Diff line number	Diff line change
`@@ -55,6 +55,7 @@ protected void subscribeActual(Observer<? super R> observer) {`
`55`	`55`	`stream = Objects.requireNonNull(mapper.apply(t), "The mapper returned a null Stream");`
`56`	`56`	`}`
`57`	`57`	`} catch (Throwable ex) {`
	`58`	`+ Exceptions.throwIfFatal(ex);`
`58`	`59`	`EmptyDisposable.error(ex, observer);`
`59`	`60`	`return;`
`60`	`61`	`}`
Original file line number	Diff line number	Diff line change
`@@ -39,6 +39,7 @@ protected void subscribeActual(Subscriber<? super U> s) {`
`39`	`39`	`try {`
`40`	`40`	`u = Objects.requireNonNull(initialSupplier.get(), "The initial value supplied is null");`
`41`	`41`	`} catch (Throwable e) {`
	`42`	`+ Exceptions.throwIfFatal(e);`
`42`	`43`	`EmptySubscription.error(e, s);`
`43`	`44`	`return;`
`44`	`45`	`}`
Original file line number	Diff line number	Diff line change
`@@ -44,6 +44,7 @@ protected void subscribeActual(SingleObserver<? super U> observer) {`
`44`	`44`	`try {`
`45`	`45`	`u = Objects.requireNonNull(initialSupplier.get(), "The initialSupplier returned a null value");`
`46`	`46`	`} catch (Throwable e) {`
	`47`	`+ Exceptions.throwIfFatal(e);`
`47`	`48`	`EmptyDisposable.error(e, observer);`
`48`	`49`	`return;`
`49`	`50`	`}`
Original file line number	Diff line number	Diff line change
`@@ -209,6 +209,7 @@ public void connect(Consumer<? super Disposable> connection) {`
`209`	`209`	`try {`
`210`	`210`	`connection.accept(ps);`
`211`	`211`	`} catch (Throwable ex) {`
	`212`	`+ Exceptions.throwIfFatal(ex);`
`212`	`213`	`if (doConnect) {`
`213`	`214`	`ps.shouldConnect.compareAndSet(true, false);`
`214`	`215`	`}`
Original file line number	Diff line number	Diff line change
`@@ -136,6 +136,7 @@ public void subscribeActual(Subscriber<? super R> s) {`
`136`	`136`	`try {`
`137`	`137`	`other = Objects.requireNonNull(mapper.apply(value), "The mapper returned a null Publisher");`
`138`	`138`	`} catch (Throwable e) {`
	`139`	`+ Exceptions.throwIfFatal(e);`
`139`	`140`	`EmptySubscription.error(e, s);`
`140`	`141`	`return;`
`141`	`142`	`}`
Original file line number	Diff line number	Diff line change
`@@ -52,6 +52,7 @@ protected void subscribeActual(MaybeObserver<? super T> observer) {`
`52`	`52`	`v = future.get(timeout, unit);`
`53`	`53`	`}`
`54`	`54`	`} catch (Throwable ex) {`
	`55`	`+ Exceptions.throwIfFatal(ex);`
`55`	`56`	`if (ex instanceof ExecutionException) {`
`56`	`57`	`ex = ex.getCause();`
`57`	`58`	`}`
Original file line number	Diff line number	Diff line change
`@@ -205,6 +205,7 @@ public void connect(Consumer<? super Disposable> connection) {`
`205`	`205`	`try {`
`206`	`206`	`connection.accept(ps);`
`207`	`207`	`} catch (Throwable ex) {`
	`208`	`+ Exceptions.throwIfFatal(ex);`
`208`	`209`	`if (doConnect) {`
`209`	`210`	`ps.shouldConnect.compareAndSet(true, false);`
`210`	`211`	`}`
Original file line number	Diff line number	Diff line change
`@@ -140,6 +140,7 @@ public void subscribeActual(Observer<? super R> observer) {`
`140`	`140`	`try {`
`141`	`141`	`other = Objects.requireNonNull(mapper.apply(value), "The mapper returned a null ObservableSource");`
`142`	`142`	`} catch (Throwable e) {`
	`143`	`+ Exceptions.throwIfFatal(e);`
`143`	`144`	`EmptyDisposable.error(e, observer);`
`144`	`145`	`return;`
`145`	`146`	`}`