2.x: Report errors from onError to Plugin when done (#4590)

ReactiveX · Sep 23, 2016 · 7791076 · 7791076
1 parent c5e38ce
commit 7791076
Show file tree

Hide file tree

Showing 23 changed files with 94 additions and 20 deletions.
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableAny.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableAny.java
@@ -12,6 +12,7 @@
  */
 package io.reactivex.internal.operators.flowable;
 
+import io.reactivex.plugins.RxJavaPlugins;
 import org.reactivestreams.*;
 
 import io.reactivex.exceptions.Exceptions;
@@ -76,10 +77,13 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable t) {
-            if (!done) {
-                done = true;
-                actual.onError(t);
+            if (done) {
+                RxJavaPlugins.onError(t);
+                return;
             }
+
+            done = true;
+            actual.onError(t);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableAnySingle.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableAnySingle.java
@@ -90,11 +90,14 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable t) {
-            if (!done) {
-                done = true;
-                s = SubscriptionHelper.CANCELLED;
-                actual.onError(t);
+            if (done) {
+                RxJavaPlugins.onError(t);
+                return;
             }
+
+            done = true;
+            s = SubscriptionHelper.CANCELLED;
+            actual.onError(t);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMap.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMap.java
@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import io.reactivex.plugins.RxJavaPlugins;
 import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.*;
 
@@ -309,6 +310,7 @@ void tryEmit(U value, InnerSubscriber<T, U> inner) {
         public void onError(Throwable t) {
             // safeguard against misbehaving sources
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             getErrorQueue().offer(t);

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureDrop.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureDrop.java
@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import io.reactivex.plugins.RxJavaPlugins;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.reactivestreams.*;
@@ -97,6 +98,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             done = true;

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableSingle.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableSingle.java
@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import io.reactivex.plugins.RxJavaPlugins;
 import org.reactivestreams.*;
 
 import io.reactivex.internal.subscriptions.*;
@@ -73,6 +74,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             done = true;

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableSingleMaybe.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableSingleMaybe.java
@@ -81,6 +81,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             done = true;

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableSingleSingle.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableSingleSingle.java
@@ -87,6 +87,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             done = true;

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableTakeWhile.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableTakeWhile.java
@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import io.reactivex.plugins.RxJavaPlugins;
 import org.reactivestreams.*;
 
 import io.reactivex.exceptions.Exceptions;
@@ -80,6 +81,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             done = true;

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableAny.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableAny.java
@@ -17,6 +17,7 @@
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.Predicate;
 import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableAny<T> extends AbstractObservableWithUpstream<T, Boolean> {
     final Predicate<? super T> predicate;
@@ -75,10 +76,13 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable t) {
-            if (!done) {
-                done = true;
-                actual.onError(t);
+            if (done) {
+                RxJavaPlugins.onError(t);
+                return;
             }
+
+            done = true;
+            actual.onError(t);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableAnySingle.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableAnySingle.java
@@ -84,10 +84,13 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable t) {
-            if (!done) {
-                done = true;
-                actual.onError(t);
+            if (done) {
+                RxJavaPlugins.onError(t);
+                return;
             }
+
+            done = true;
+            actual.onError(t);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableElementAt.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableElementAt.java
@@ -16,6 +16,7 @@
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableElementAt<T> extends AbstractObservableWithUpstream<T, T> {
     final long index;
@@ -86,6 +87,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             done = true;

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableElementAtMaybe.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableElementAtMaybe.java
@@ -89,6 +89,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             done = true;

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableElementAtSingle.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableElementAtSingle.java
@@ -16,6 +16,7 @@
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableElementAtSingle<T> extends Single<T> {
     final ObservableSource<T> source;
@@ -86,6 +87,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             done = true;

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMap.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMap.java
@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.operators.observable;
 
+import io.reactivex.plugins.RxJavaPlugins;
 import java.util.*;
 import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.*;
@@ -277,8 +278,8 @@ void tryEmit(U value, InnerObserver<T, U> inner) {
 
         @Override
         public void onError(Throwable t) {
-            // safeguard against misbehaving sources
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             getErrorQueue().offer(t);
@@ -288,7 +289,6 @@ public void onError(Throwable t) {
 
         @Override
         public void onComplete() {
-            // safeguard against misbehaving sources
             if (done) {
                 return;
             }

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableSingle.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableSingle.java
@@ -16,6 +16,7 @@
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableSingle<T> extends AbstractObservableWithUpstream<T, T> {
 
@@ -82,6 +83,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             done = true;

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableSingleMaybe.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableSingleMaybe.java
@@ -16,6 +16,7 @@
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableSingleMaybe<T> extends Maybe<T> {
 
@@ -79,6 +80,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             done = true;

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableSingleSingle.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableSingleSingle.java
@@ -16,6 +16,7 @@
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableSingleSingle<T> extends Single<T> {
 
@@ -85,6 +86,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             done = true;

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableTake.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableTake.java
@@ -16,6 +16,7 @@
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.*;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableTake<T> extends AbstractObservableWithUpstream<T, T> {
     final long limit;
@@ -66,11 +67,14 @@ public void onNext(T t) {
         }
         @Override
         public void onError(Throwable t) {
-            if (!done) {
-                done = true;
-                subscription.dispose();
-                actual.onError(t);
+            if (done) {
+                RxJavaPlugins.onError(t);
+                return;
             }
+
+            done = true;
+            subscription.dispose();
+            actual.onError(t);
         }
         @Override
         public void onComplete() {

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableTakeWhile.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableTakeWhile.java
@@ -18,6 +18,7 @@
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.Predicate;
 import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableTakeWhile<T> extends AbstractObservableWithUpstream<T, T> {
     final Predicate<? super T> predicate;
@@ -92,6 +93,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             done = true;

diff --git a/src/main/java/io/reactivex/observers/SafeObserver.java b/src/main/java/io/reactivex/observers/SafeObserver.java
@@ -136,6 +136,7 @@ void onNextNoSubscription() {
     @Override
     public void onError(Throwable t) {
         if (done) {
+            RxJavaPlugins.onError(t);
             return;
         }
         done = true;

diff --git a/src/main/java/io/reactivex/subjects/UnicastSubject.java b/src/main/java/io/reactivex/subjects/UnicastSubject.java
@@ -13,6 +13,7 @@
 
 package io.reactivex.subjects;
 
+import io.reactivex.plugins.RxJavaPlugins;
 import java.util.concurrent.atomic.*;
 
 import io.reactivex.Observer;
@@ -193,6 +194,7 @@ public void onNext(T t) {
     @Override
     public void onError(Throwable t) {
         if (done || disposed) {
+            RxJavaPlugins.onError(t);
             return;
         }
         if (t == null) {

diff --git a/src/main/java/io/reactivex/subscribers/SafeSubscriber.java b/src/main/java/io/reactivex/subscribers/SafeSubscriber.java
@@ -124,6 +124,7 @@ void onNextNoSubscription() {
     @Override
     public void onError(Throwable t) {
         if (done) {
+            RxJavaPlugins.onError(t);
             return;
         }
         done = true;