OnErrorResume Backpressure Tests

Author: benjchristensen

src/test/java/rx/internal/operators/OperatorOnErrorResumeNextViaFunctionTest.java

@@ -309,4 +309,39 @@ public void run() {
         }
 
     }
+    
+    @Test
+    public void testBackpressure() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        Observable.range(0, 100000)
+                .onErrorResumeNext(new Func1<Throwable, Observable<Integer>>() {
+
+                    @Override
+                    public Observable<Integer> call(Throwable t1) {
+                        return 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();
+    }
 }

src/test/java/rx/internal/operators/OperatorOnErrorResumeNextViaObservableTest.java

@@ -21,6 +21,8 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
+import java.util.concurrent.TimeUnit;
+
 import org.junit.Test;
 import org.mockito.Mockito;
 
@@ -29,6 +31,8 @@
 import rx.Subscriber;
 import rx.Subscription;
 import rx.functions.Func1;
+import rx.observers.TestSubscriber;
+import rx.schedulers.Schedulers;
 
 public class OperatorOnErrorResumeNextViaObservableTest {
 
@@ -143,4 +147,32 @@ public void run() {
             System.out.println("done starting TestObservable thread");
         }
     }
+    
+    @Test
+    public void testBackpressure() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        Observable.range(0, 100000)
+                .onErrorResumeNext(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();
+    }
 }