Merge pull request #3689 from akarnokd/OnErrorDoXFix1x

1.x: unified onErrorX and onExceptionResumeNext and fixed backpressure

Author: akarnokd

src/main/java/rx/Observable.java

@@ -6230,7 +6230,7 @@ public final Observable<T> onErrorResumeNext(final Func1<Throwable, ? extends Ob
      * @see <a href="http://reactivex.io/documentation/operators/catch.html">ReactiveX operators documentation: Catch</a>
      */
     public final Observable<T> onErrorResumeNext(final Observable<? extends T> resumeSequence) {
-        return lift(new OperatorOnErrorResumeNextViaObservable<T>(resumeSequence));
+        return lift(OperatorOnErrorResumeNextViaFunction.withOther(resumeSequence));
     }
 
     /**
@@ -6260,7 +6260,7 @@ public final Observable<T> onErrorResumeNext(final Observable<? extends T> resum
      * @see <a href="http://reactivex.io/documentation/operators/catch.html">ReactiveX operators documentation: Catch</a>
      */
     public final Observable<T> onErrorReturn(Func1<Throwable, ? extends T> resumeFunction) {
-        return lift(new OperatorOnErrorReturn<T>(resumeFunction));
+        return lift(OperatorOnErrorResumeNextViaFunction.withSingle(resumeFunction));
     }
 
     /**
@@ -6296,7 +6296,7 @@ public final Observable<T> onErrorReturn(Func1<Throwable, ? extends T> resumeFun
      * @see <a href="http://reactivex.io/documentation/operators/catch.html">ReactiveX operators documentation: Catch</a>
      */
     public final Observable<T> onExceptionResumeNext(final Observable<? extends T> resumeSequence) {
-        return lift(new OperatorOnExceptionResumeNextViaObservable<T>(resumeSequence));
+        return lift(OperatorOnErrorResumeNextViaFunction.withException(resumeSequence));
     }
 
     /**

src/main/java/rx/Single.java

@@ -1411,7 +1411,7 @@ public final Single<T> observeOn(Scheduler scheduler) {
      * @see <a href="http://reactivex.io/documentation/operators/catch.html">ReactiveX operators documentation: Catch</a>
      */
     public final Single<T> onErrorReturn(Func1<Throwable, ? extends T> resumeFunction) {
-        return lift(new OperatorOnErrorReturn<T>(resumeFunction));
+        return lift(OperatorOnErrorResumeNextViaFunction.withSingle(resumeFunction));
     }
 
     /**

src/main/java/rx/internal/operators/OperatorOnErrorResumeNextViaFunction.java

@@ -45,17 +45,51 @@ public final class OperatorOnErrorResumeNextViaFunction<T> implements Operator<T
 
     final Func1<Throwable, ? extends Observable<? extends T>> resumeFunction;
 
+    public static <T> OperatorOnErrorResumeNextViaFunction<T> withSingle(final Func1<Throwable, ? extends T> resumeFunction) {
+        return new OperatorOnErrorResumeNextViaFunction<T>(new Func1<Throwable, Observable<? extends T>>() {
+            @Override
+            public Observable<? extends T> call(Throwable t) {
+                return Observable.just(resumeFunction.call(t));
+            }
+        });
+    }
+
+    public static <T> OperatorOnErrorResumeNextViaFunction<T> withOther(final Observable<? extends T> other) {
+        return new OperatorOnErrorResumeNextViaFunction<T>(new Func1<Throwable, Observable<? extends T>>() {
+            @Override
+            public Observable<? extends T> call(Throwable t) {
+                return other;
+            }
+        });
+    }
+
+    public static <T> OperatorOnErrorResumeNextViaFunction<T> withException(final Observable<? extends T> other) {
+        return new OperatorOnErrorResumeNextViaFunction<T>(new Func1<Throwable, Observable<? extends T>>() {
+            @Override
+            public Observable<? extends T> call(Throwable t) {
+                if (t instanceof Exception) {
+                    return other;
+                }
+                return Observable.error(t);
+            }
+        });
+    }
+
     public OperatorOnErrorResumeNextViaFunction(Func1<Throwable, ? extends Observable<? extends T>> f) {
         this.resumeFunction = f;
     }
 
     @Override
     public Subscriber<? super T> call(final Subscriber<? super T> child) {
         final ProducerArbiter pa = new ProducerArbiter();
+        
         final SerialSubscription ssub = new SerialSubscription();
+        
         Subscriber<T> parent = new Subscriber<T>() {
 
-            private boolean done = false;
+            private boolean done;
+        
+            long produced;
 
             @Override
             public void onCompleted() {
@@ -70,12 +104,13 @@ public void onCompleted() {
             public void onError(Throwable e) {
                 if (done) {
                     Exceptions.throwIfFatal(e);
+                    RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
                     return;
                 }
                 done = true;
                 try {
-                    RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
                     unsubscribe();
+
                     Subscriber<T> next = new Subscriber<T>() {
                         @Override
                         public void onNext(T t) {
@@ -96,7 +131,13 @@ public void setProducer(Producer producer) {
                     };
                     ssub.set(next);
 
+                    long p = produced;
+                    if (p != 0L) {
+                        pa.produced(p);
+                    }
+                    
                     Observable<? extends T> resume = resumeFunction.call(e);
+                    
                     resume.unsafeSubscribe(next);
                 } catch (Throwable e2) {
                     Exceptions.throwOrReport(e2, child);
@@ -108,6 +149,7 @@ public void onNext(T t) {
                 if (done) {
                     return;
                 }
+                produced++;
                 child.onNext(t);
             }
 
@@ -117,9 +159,11 @@ public void setProducer(final Producer producer) {
             }
 
         };
-        child.add(ssub);
         ssub.set(parent);
+
+        child.add(ssub);
         child.setProducer(pa);
+        
         return parent;
     }