Merge pull request #2563 from akarnokd/RetryBackpressureParallelTestChange

Retry backpressure test: split error conditions into separate test lines.

Author: akarnokd

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

@@ -15,33 +15,22 @@
  */
 package rx.internal.operators;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
+import static org.junit.Assert.*;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.*;
 
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
 
 import org.junit.Test;
-import org.mockito.InOrder;
-import org.mockito.Mockito;
+import org.mockito.*;
 
-import rx.Observable;
+import rx.*;
 import rx.Observable.OnSubscribe;
+import rx.Observable;
 import rx.Observer;
-import rx.Producer;
-import rx.Subscriber;
-import rx.Subscription;
-import rx.functions.Action0;
-import rx.functions.Action1;
-import rx.functions.Func1;
-import rx.functions.Func2;
+import rx.functions.*;
 import rx.internal.util.RxRingBuffer;
 import rx.observables.GroupedObservable;
 import rx.observers.TestSubscriber;
@@ -722,7 +711,10 @@ public void testRetryWithBackpressureParallel() throws InterruptedException {
         int ncpu = Runtime.getRuntime().availableProcessors();
         ExecutorService exec = Executors.newFixedThreadPool(Math.max(ncpu / 2, 1));
         final AtomicInteger timeouts = new AtomicInteger();
-        final AtomicInteger data = new AtomicInteger();
+        final Map<Integer, List<String>> data = new ConcurrentHashMap<Integer, List<String>>();
+        final Map<Integer, List<Throwable>> exceptions = new ConcurrentHashMap<Integer, List<Throwable>>();
+        final Map<Integer, Integer> completions = new ConcurrentHashMap<Integer, Integer>();
+        
         int m = 2000;
         final CountDownLatch cdl = new CountDownLatch(m);
         for (int i = 0; i < m; i++) {
@@ -737,13 +729,13 @@ public void run() {
                         origin.retry().observeOn(Schedulers.computation()).unsafeSubscribe(ts);
                         ts.awaitTerminalEvent(2, TimeUnit.SECONDS);
                         if (ts.getOnNextEvents().size() != NUM_RETRIES + 2) {
-                            data.incrementAndGet();
+                            data.put(j, ts.getOnNextEvents());
                         }
                         if (ts.getOnErrorEvents().size() != 0) {
-                            data.incrementAndGet();
+                            exceptions.put(j, ts.getOnErrorEvents());
                         }
                         if (ts.getOnCompletedEvents().size() != 1) {
-                            data.incrementAndGet();
+                            completions.put(j, ts.getOnCompletedEvents().size());
                         }
                     } catch (Throwable t) {
                         timeouts.incrementAndGet();
@@ -756,8 +748,15 @@ public void run() {
         exec.shutdown();
         cdl.await();
         assertEquals(0, timeouts.get());
-        assertEquals(0, data.get());
-
+        if (data.size() > 0) {
+            fail("Data content mismatch: " + data);
+        }
+        if (exceptions.size() > 0) {
+            fail("Exceptions received: " + exceptions);
+        }
+        if (completions.size() > 0) {
+            fail("Multiple completions received: " + completions);
+        }
     }
     @Test(timeout = 3000)
     public void testIssue1900() throws InterruptedException {