Merge pull request #1771 from neerajrj/onErrorReturnBackpressure

On error return backpressure
ReactiveX · Oct 16, 2014 · 502405d · 502405d
2 parents 8ed945f + 0b01c02
commit 502405d
Show file tree

Hide file tree

Showing 4 changed files with 93 additions and 0 deletions.
diff --git a/src/main/java/rx/internal/operators/OperatorOnErrorReturn.java b/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;

diff --git a/src/main/java/rx/internal/operators/OperatorOnExceptionResumeNextViaObservable.java b/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);
 

diff --git a/src/test/java/rx/internal/operators/OperatorOnErrorReturnTest.java b/src/test/java/rx/internal/operators/OperatorOnErrorReturnTest.java
@@ -32,6 +32,7 @@
 import rx.Subscriber;
 import rx.functions.Func1;
 import rx.observers.TestSubscriber;
+import rx.schedulers.Schedulers;
 
 public class OperatorOnErrorReturnTest {
 
@@ -145,6 +146,41 @@ public String call(Throwable t1) {
         verify(observer, Mockito.never()).onNext("three");
         verify(observer, times(1)).onNext("resume");
     }
+
+    @Test
+    public void testBackpressure() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        Observable.range(0, 100000)
+                .onErrorReturn(new Func1<Throwable, Integer>() {
+
+					@Override
+					public Integer call(Throwable t1) {
+						return 1;
+					}
+
+                })
+                .observeOn(Schedulers.computation())
+                .map(new Func1<Integer, Integer>() {
+                    int c = 0;
+
+                    @Override
+                    public Integer call(Integer t1) {
+                        if (c++ <= 1) {
+                            // slow
+                            try {
+                                Thread.sleep(500);
+                            } catch (InterruptedException e) {
+                                e.printStackTrace();
+                            }
+                        }
+                        return t1;
+                    }
+
+                })
+                .subscribe(ts);
+        ts.awaitTerminalEvent();
+        ts.assertNoErrors();
+    }
 
     private static class TestObservable implements Observable.OnSubscribe<String> {
 
@@ -180,4 +216,7 @@ public void run() {
             System.out.println("done starting TestObservable thread");
         }
     }
+
+
+
 }
diff --git a/src/test/java/rx/internal/operators/OperatorOnExceptionResumeNextViaObservableTest.java b/src/test/java/rx/internal/operators/OperatorOnExceptionResumeNextViaObservableTest.java
@@ -30,6 +30,8 @@
 import rx.Observer;
 import rx.Subscriber;
 import rx.functions.Func1;
+import rx.observers.TestSubscriber;
+import rx.schedulers.Schedulers;
 
 public class OperatorOnExceptionResumeNextViaObservableTest {
 
@@ -188,6 +190,36 @@ public String call(String s) {
         verify(observer, Mockito.never()).onError(any(Throwable.class));
         verify(observer, times(1)).onCompleted();
     }
+
+
+    @Test
+    public void testBackpressure() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        Observable.range(0, 100000)
+                .onExceptionResumeNext(Observable.just(1))
+                .observeOn(Schedulers.computation())
+                .map(new Func1<Integer, Integer>() {
+                    int c = 0;
+
+                    @Override
+                    public Integer call(Integer t1) {
+                        if (c++ <= 1) {
+                            // slow
+                            try {
+                                Thread.sleep(500);
+                            } catch (InterruptedException e) {
+                                e.printStackTrace();
+                            }
+                        }
+                        return t1;
+                    }
+
+                })
+                .subscribe(ts);
+        ts.awaitTerminalEvent();
+        ts.assertNoErrors();
+    }
+
 
     private static class TestObservable implements Observable.OnSubscribe<String> {