Merge pull request #731 from zsxwing/flaky-unit-tests

Replaced 'Thread.sleep' with 'CountDownLatch' to fix the flaky test failures

Author: benjchristensen

rxjava-core/src/test/java/rx/operators/OperationConcatTest.java

@@ -156,6 +156,7 @@ public void testNestedAsyncConcat() throws Throwable {
         final CountDownLatch allowThird = new CountDownLatch(1);
 
         final AtomicReference<Thread> parent = new AtomicReference<Thread>();
+        final CountDownLatch parentHasStarted = new CountDownLatch(1);
         Observable<Observable<String>> observableOfObservables = Observable.create(new Observable.OnSubscribeFunc<Observable<String>>() {
 
             @Override
@@ -197,29 +198,22 @@ public void run() {
                     }
                 }));
                 parent.get().start();
+                parentHasStarted.countDown();
                 return s;
             }
         });
 
         Observable.create(concat(observableOfObservables)).subscribe(observer);
 
         // wait for parent to start
-        while (parent.get() == null) {
-            Thread.sleep(1);
-        }
+        parentHasStarted.await();
 
         try {
             // wait for first 2 async observables to complete
-            while (o1.t == null) {
-                Thread.sleep(1);
-            }
-            System.out.println("Thread1 started ... waiting for it to complete ...");
-            o1.t.join();
-            while (o2.t == null) {
-                Thread.sleep(1);
-            }
-            System.out.println("Thread2 started ... waiting for it to complete ...");
-            o2.t.join();
+            System.out.println("Thread1 is starting ... waiting for it to complete ...");
+            o1.waitForThreadDone();
+            System.out.println("Thread2 is starting ... waiting for it to complete ...");
+            o2.waitForThreadDone();
         } catch (Throwable e) {
             throw new RuntimeException("failed waiting on threads", e);
         }
@@ -243,11 +237,8 @@ public void run() {
         allowThird.countDown();
 
         try {
-            while (o3.t == null) {
-                Thread.sleep(1);
-            }
             // wait for 3rd to complete
-            o3.t.join();
+            o3.waitForThreadDone();
         } catch (Throwable e) {
             throw new RuntimeException("failed waiting on threads", e);
         }
@@ -320,9 +311,8 @@ public void testConcatConcurrentWithInfinity() {
 
         //Wait for the thread to start up.
         try {
-            Thread.sleep(25);
-            w1.t.join();
-            w2.t.join();
+            w1.waitForThreadDone();
+            w2.waitForThreadDone();
         } catch (InterruptedException e) {
             // TODO Auto-generated catch block
             e.printStackTrace();
@@ -500,6 +490,7 @@ public void unsubscribe() {
         private boolean subscribed = true;
         private final CountDownLatch once;
         private final CountDownLatch okToContinue;
+        private final CountDownLatch threadHasStarted = new CountDownLatch(1);
         private final T seed;
         private final int size;
 
@@ -553,8 +544,14 @@ public void run() {
 
             });
             t.start();
+            threadHasStarted.countDown();
             return s;
         }
+
+        void waitForThreadDone() throws InterruptedException {
+            threadHasStarted.await();
+            t.join();
+        }
     }
     @Test
     public void testMultipleObservers() {