Added 'sleep' to avoid 'testMapWithErrorInFuncAndThreadPoolScheduler' fails

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

@@ -15,25 +15,35 @@
  */
 package rx.operators;
 
-import static org.junit.Assert.*;
-import static org.mockito.Matchers.*;
-import static org.mockito.Mockito.*;
-import static rx.operators.OperationMap.*;
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static rx.operators.OperationMap.map;
+import static rx.operators.OperationMap.mapMany;
+import static rx.operators.OperationMap.mapWithIndex;
 
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.InOrder;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
 import rx.Observable;
 import rx.Observer;
 import rx.concurrency.Schedulers;
+import rx.concurrency.TestScheduler;
 import rx.util.functions.Func1;
 import rx.util.functions.Func2;
 
@@ -258,6 +268,27 @@ public Integer call(Integer arg0) {
                 }).toBlockingObservable().single();
     }
 
+    @Test
+    public void testMapWithErrorInFuncAndScheduler() throws InterruptedException {
+        // The error will throw in the scheduler.
+        // map needs to handle the error and notice the observer.
+        TestScheduler scheduler = new TestScheduler();
+        Observable<String> m = Observable.from("one")
+                .observeOn(scheduler)
+                .map(new Func1<String, String>() {
+                    public String call(String arg0) {
+                        throw new IllegalArgumentException("any error");
+                    }
+                });
+
+        m.subscribe(stringObserver);
+
+        scheduler.advanceTimeBy(1000, TimeUnit.MILLISECONDS);
+        InOrder inorder = inOrder(stringObserver);
+        inorder.verify(stringObserver, times(1)).onError(any(IllegalArgumentException.class));
+        inorder.verifyNoMoreInteractions();
+    }
+
     @Test
     public void testMapWithErrorInFuncAndThreadPoolScheduler() throws InterruptedException {
         // The error will throw in one of threads in the thread pool.
@@ -268,14 +299,18 @@ public void testMapWithErrorInFuncAndThreadPoolScheduler() throws InterruptedExc
                 .observeOn(Schedulers.threadPoolForComputation())
                 .map(new Func1<String, String>() {
                     public String call(String arg0) {
-                        try {
-                            throw new IllegalArgumentException("any error");
-                        } finally {
-                            latch.countDown();
-                        }
+                        throw new IllegalArgumentException("any error");
                     }
                 });
 
+        // wait for the call to get to the observer before decrementing the latch
+        doAnswer(new Answer<Void>() {
+            @Override
+            public Void answer(InvocationOnMock invocation) throws Throwable {
+                latch.countDown();
+                return null;
+            }
+        }).when(stringObserver).onError(any(Throwable.class));
         m.subscribe(stringObserver);
         latch.await();
         InOrder inorder = inOrder(stringObserver);