Merge pull request #1765 from neerajrj/onErrorResumeBackpressure

backpressure support in onErrorResumeNext* operators

Author: benjchristensen

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

@@ -16,6 +16,7 @@
 package rx.internal.operators;
 
 import rx.Observable;
+import rx.Producer;
 import rx.Observable.Operator;
 import rx.Subscriber;
 import rx.exceptions.Exceptions;
@@ -87,6 +88,16 @@ public void onNext(T t) {
                 }
                 child.onNext(t);
             }
+            
+            @Override
+            public void setProducer(final Producer producer) {
+                child.setProducer(new Producer() {
+                    @Override
+                    public void request(long n) {
+                        producer.request(n);
+                    }
+                });
+            }
 
         };
         child.add(parent);

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

@@ -16,6 +16,7 @@
 package rx.internal.operators;
 
 import rx.Observable;
+import rx.Producer;
 import rx.Observable.Operator;
 import rx.Subscriber;
 import rx.exceptions.Exceptions;
@@ -84,6 +85,16 @@ public void onCompleted() {
                 child.onCompleted();
             }
 
+            @Override
+            public void setProducer(final Producer producer) {
+                child.setProducer(new Producer() {
+                    @Override
+                    public void request(long n) {
+                        producer.request(n);
+                    }
+                });
+            }
+            
         };
         child.add(s);