back pressure tests

Author: neerajrj

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");
         }
     }
+    
+    
+    
 }

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> {