back pressure support

Author: neerajrj

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

@@ -18,6 +18,7 @@
 import java.util.Arrays;
 
 import rx.Observable.Operator;
+import rx.Producer;
 import rx.Subscriber;
 import rx.exceptions.CompositeException;
 import rx.exceptions.Exceptions;
@@ -92,6 +93,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(parent);
         return parent;

src/main/java/rx/internal/operators/OperatorOnExceptionResumeNextViaObservable.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;
@@ -92,6 +93,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);