Merge pull request #777 from akarnokd/NextTestFix

Fixed testSingleSourceManyIterators
ReactiveX · Feb 11, 2014 · 485c22a · 485c22a
2 parents 0d83c94 + 0202e63
commit 485c22a
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 9 deletions.
diff --git a/rxjava-core/src/main/java/rx/operators/OperationNext.java b/rxjava-core/src/main/java/rx/operators/OperationNext.java
@@ -48,7 +48,8 @@ public Iterator<T> iterator() {
 
     }
 
-    private static class NextIterator<T> implements Iterator<T> {
+    // test needs to access the observer.waiting flag non-blockingly.
+    /* private */static final class NextIterator<T> implements Iterator<T> {
 
         private final NextObserver<? extends T> observer;
         private T next;
@@ -60,6 +61,12 @@ private NextIterator(NextObserver<? extends T> observer) {
             this.observer = observer;
         }
 
+        // in tests, set the waiting flag without blocking for the next value to 
+        // allow lockstepping instead of multi-threading
+        void setWaiting(boolean value) {
+            observer.waiting.set(value);
+        }
+
         @Override
         public boolean hasNext() {
             if (error != null) {

diff --git a/rxjava-core/src/test/java/rx/operators/OperationNextTest.java b/rxjava-core/src/test/java/rx/operators/OperationNextTest.java
@@ -296,24 +296,27 @@ public void run() {
         System.out.println("a: " + a + " b: " + b + " c: " + c);
     }
 
-    @Test(timeout = 8000)
+    @Test /* (timeout = 8000) */
     public void testSingleSourceManyIterators() throws InterruptedException {
-        BlockingObservable<Long> source = Observable.interval(200, TimeUnit.MILLISECONDS).take(10).toBlockingObservable();
+        PublishSubject<Long> ps = PublishSubject.create();
+        BlockingObservable<Long> source = ps.take(10).toBlockingObservable();
 
         Iterable<Long> iter = source.next();
 
         for (int j = 0; j < 3; j++) {
-            Iterator<Long> it = iter.iterator();
+            OperationNext.NextIterator<Long> it = (OperationNext.NextIterator<Long>)iter.iterator();
 
-            for (int i = 0; i < 9; i++) {
+            for (long i = 0; i < 9; i++) {
                 // hasNext has to set the waiting to true, otherwise, all onNext will be skipped
+                it.setWaiting(true);
+                ps.onNext(i);
                 Assert.assertEquals(true, it.hasNext());
-                Assert.assertEquals(Long.valueOf(i), it.next());
+                Assert.assertEquals(j + "th iteration", Long.valueOf(i), it.next());
             }
+            it.setWaiting(true);
+            ps.onNext(9L);
 
-            Thread.sleep(400);
-
-            Assert.assertEquals(false, it.hasNext());
+            Assert.assertEquals(j + "th iteration", false, it.hasNext());
         }
 
     }