1.x: SyncOnSubscribeTest.testConcurrentRequests give more time.

The test failed on Travis and locally if my machine was under heavy load
without interacting with the mock. This change gives more time in the
inner await and reports the exception instead of itself throwing.

Author: akarnokd

src/test/java/rx/observables/SyncOnSubscribeTest.java

@@ -16,57 +16,25 @@
 
 package rx.observables;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.isA;
-import static org.mockito.Mockito.inOrder;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.BrokenBarrierException;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.CyclicBarrier;
-import java.util.concurrent.ExecutionException;
-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;
-import java.util.concurrent.atomic.AtomicReference;
+import static org.junit.Assert.*;
+import static org.mockito.Matchers.*;
+import static org.mockito.Mockito.*;
+
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
 
 import org.junit.Test;
-import org.mockito.InOrder;
-import org.mockito.Matchers;
-import org.mockito.Mockito;
+import org.mockito.*;
 
+import rx.*;
 import rx.Observable;
-import rx.Observable.OnSubscribe;
-import rx.Observable.Operator;
+import rx.Observable.*;
 import rx.Observer;
-import rx.Producer;
-import rx.Subscriber;
 import rx.exceptions.TestException;
-import rx.functions.Action0;
-import rx.functions.Action1;
-import rx.functions.Action2;
-import rx.functions.Func0;
-import rx.functions.Func2;
+import rx.functions.*;
 import rx.observers.TestSubscriber;
-import rx.schedulers.Schedulers;
-import rx.schedulers.TestScheduler;
+import rx.schedulers.*;
 
 /**
  * Test if SyncOnSubscribe adheres to the usual unsubscription and backpressure contracts.
@@ -489,6 +457,16 @@ public Integer call(Integer state, Observer<? super Integer> observer) {
         verify(onUnSubscribe, times(1)).call(any(Integer.class));
     }
 
+    @Test
+    public void testConcurrentRequestsLoop() throws InterruptedException {
+        for (int i = 0; i < 100; i++) {
+            if (i % 10 == 0) {
+                System.out.println("testConcurrentRequestsLoop >> " + i);
+            }
+            testConcurrentRequests();
+        }
+    }
+    
     @Test
     public void testConcurrentRequests() throws InterruptedException {
         final int count1 = 1000;
@@ -514,12 +492,20 @@ public Integer call(Integer state, Observer<? super Integer> observer) {
                         l2.countDown();
                         // wait until the 2nd request returns then proceed
                         try {
-                            if (!l1.await(1, TimeUnit.SECONDS))
-                                throw new IllegalStateException();
-                        } catch (InterruptedException e) {}
+                            if (!l1.await(2, TimeUnit.SECONDS)) {
+                                observer.onError(new TimeoutException());
+                                return state + 1;
+                            }
+                        } catch (InterruptedException e) {
+                            observer.onError(e);
+                            return state + 1;
+                        }
                         observer.onNext(state);
-                        if (state == finalCount)
+                        
+                        if (state == finalCount) {
                             observer.onCompleted();
+                        }
+                        
                         return state + 1;
                     }},
                 onUnSubscribe);
@@ -532,10 +518,9 @@ public Integer call(Integer state, Observer<? super Integer> observer) {
         Observable.create(os).subscribeOn(Schedulers.newThread()).subscribe(ts);
 
         // wait until the first request has started processing
-        try {
-            if (!l2.await(1, TimeUnit.SECONDS))
-                throw new IllegalStateException();
-        } catch (InterruptedException e) {}
+        if (!l2.await(2, TimeUnit.SECONDS)) {
+            fail("SyncOnSubscribe failed to countDown in time");
+        }
         // make a concurrent request, this should return
         ts.requestMore(count2);
         // unblock the 1st thread to proceed fulfilling requests