From e2c379c575885099b252d599072e426689071742 Mon Sep 17 00:00:00 2001
From: Ben Christensen <benjchristensen@netflix.com>
Date: Sat, 31 Aug 2013 13:54:17 -0700
Subject: [PATCH] Improve testNoBufferingOrBlockingOfSequence determinism

Removed use of Thread.sleep to fix non-determinism issue reported in https://github.com/Netflix/RxJava/issues/329
---
 .../main/java/rx/operators/OperationNext.java    | 16 +++++++---------
 1 file changed, 7 insertions(+), 9 deletions(-)

diff --git a/rxjava-core/src/main/java/rx/operators/OperationNext.java b/rxjava-core/src/main/java/rx/operators/OperationNext.java
index 587141b1eb3..5c0a8a75538 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationNext.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationNext.java
@@ -27,6 +27,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -313,6 +314,8 @@ private static class TestException extends RuntimeException {
         @Test
         public void testNoBufferingOrBlockingOfSequence() throws Throwable {
             final CountDownLatch finished = new CountDownLatch(1);
+            final int COUNT = 30;
+            final CountDownLatch timeHasPassed = new CountDownLatch(COUNT);
             final AtomicBoolean running = new AtomicBoolean(true);
             final AtomicInteger count = new AtomicInteger(0);
             final Observable<Integer> obs = Observable.create(new Func1<Observer<Integer>, Subscription>() {
@@ -326,7 +329,7 @@ public void run() {
                             try {
                                 while (running.get()) {
                                     o.onNext(count.incrementAndGet());
-                                    Thread.sleep(0, 100);
+                                    timeHasPassed.countDown();
                                 }
                                 o.onCompleted();
                             } catch (Throwable e) {
@@ -350,19 +353,14 @@ public void run() {
             // we should have a different value
             assertTrue("a and b should be different", a != b);
 
-            // wait for some time
-            Thread.sleep(100);
-            // make sure the counter in the observable has increased beyond b
-            while (count.get() <= (b + 10)) {
-                Thread.sleep(100);
-            }
+            // wait for some time (if times out we are blocked somewhere so fail ... set very high for very slow, constrained machines)
+            timeHasPassed.await(8000, TimeUnit.MILLISECONDS);
 
             assertTrue(it.hasNext());
-            int expectedHigherThan = count.get();
             int c = it.next();
 
             assertTrue("c should not just be the next in sequence", c != (b + 1));
-            assertTrue("expected that c [" + c + "] is higher than " + expectedHigherThan, c > expectedHigherThan);
+            assertTrue("expected that c [" + c + "] is higher than or equal to " + COUNT, c >= COUNT);
 
             assertTrue(it.hasNext());