test with observeOn

Author: abersnaze

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

@@ -19,15 +19,21 @@
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.InOrder;
 
 import rx.Observable;
 import rx.Producer;
+import rx.Scheduler;
 import rx.Subscriber;
 import rx.functions.Action1;
+import rx.functions.Func1;
 import rx.observers.TestSubscriber;
+import rx.schedulers.Schedulers;
 import rx.subjects.PublishSubject;
 
 public class OperatorValveTest {
@@ -131,6 +137,14 @@ public void test() {
         order.verifyNoMoreInteractions();
     }
 
+    @Test
+    public void testRequestError() {
+        TestSubscriber<Void> tSub = TestSubscriber.create(-1);
+        Exception e = new Exception();
+        Observable.<Void> error(e).pressureValve(Observable.<Boolean> never(), 10).subscribe(tSub);
+        tSub.assertError(e);
+    }
+
     @Test
     public void testDataError() {
         TestSubscriber<Void> tSub = TestSubscriber.create();
@@ -160,4 +174,32 @@ public void testControlCompleteClosed() {
         Observable.<Void> never().pressureValve(Observable.just(false), 10).subscribe(tSub);
         tSub.assertError(IllegalStateException.class);
     }
+
+    /*
+    @Test
+    public void testObserveOn() {
+        final AtomicLong counter = new AtomicLong();
+        Observable<Integer> range = Observable.range(0, Integer.MAX_VALUE);
+        Observable<Boolean> control = Observable.interval(1, TimeUnit.SECONDS).map(new Func1<Long, Boolean>() {
+            @Override
+            public Boolean call(Long i) {
+                System.out.println();
+                counter.set(0);
+                return i % 2 == 1;
+            }
+        });
+        long granularity = 10;
+        TestSubscriber<Integer> tSub = new TestSubscriber<Integer>();
+        range.pressureValve(control, granularity).observeOn(Schedulers.computation()).toBlocking().forEach(new Action1<Integer>() {
+            @Override
+            public void call(Integer t) {
+                System.out.print(counter.incrementAndGet()+ "         \r");
+            }
+        });
+    }
+    
+    public static void main(String[] args) {
+        new OperatorValveTest().testObserveOn();
+    }
+    */
 }