Unbroken TestSubscriber

Author: akarnokd

src/main/java/rx/observers/TestSubscriber.java

@@ -16,12 +16,9 @@
 package rx.observers;
 
 import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 
-import rx.Notification;
-import rx.Observer;
-import rx.Subscriber;
+import rx.*;
 
 /**
  * A {@code TestSubscriber} is a variety of {@link Subscriber} that you can use for unit testing, to perform
@@ -227,9 +224,11 @@ public void awaitTerminalEvent() {
      * @throws RuntimeException
      *          if the Subscriber is interrupted before the Observable is able to complete
      */
-    public boolean awaitTerminalEvent(long timeout, TimeUnit unit) {
+    public void awaitTerminalEvent(long timeout, TimeUnit unit) {
         try {
-            return latch.await(timeout, unit);
+            if (!latch.await(timeout, unit)) {
+                throw new RuntimeException(new TimeoutException());
+            }
         } catch (InterruptedException e) {
             throw new RuntimeException("Interrupted", e);
         }

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

@@ -730,15 +730,12 @@ public void testRetryWithBackpressureParallel() throws InterruptedException {
             exec.execute(new Runnable() {
                 @Override
                 public void run() {
+                    final AtomicInteger nexts = new AtomicInteger();
                     try {
-                        final AtomicInteger nexts = new AtomicInteger();
                         Observable<String> origin = Observable.create(new FuncWithErrors(NUM_RETRIES));
                         TestSubscriber<String> ts = new TestSubscriber<String>();
                         origin.retry().observeOn(Schedulers.computation()).unsafeSubscribe(ts);
-                        if (!ts.awaitTerminalEvent(2, TimeUnit.SECONDS)) {
-                            timeouts.incrementAndGet();
-                            System.out.println(j + " | " + cdl.getCount() + " !!! " + nexts.get());
-                        } 
+                        ts.awaitTerminalEvent(2, TimeUnit.SECONDS);
                         if (ts.getOnNextEvents().size() != NUM_RETRIES + 2) {
                             data.incrementAndGet();
                         }
@@ -750,6 +747,7 @@ public void run() {
                         }
                     } catch (Throwable t) {
                         timeouts.incrementAndGet();
+                        System.out.println(j + " | " + cdl.getCount() + " !!! " + nexts.get());
                     }
                     cdl.countDown();
                 }