Added a unit test to demonstrate regression in Observable.mergeDelayError

Author: Matt Jacobs

rxjava-core/src/test/java/rx/internal/operators/OperatorMergeDelayErrorTest.java

@@ -15,33 +15,27 @@
  */
 package rx.internal.operators;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.fail;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyInt;
-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.List;
-
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.InOrder;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
-
 import rx.Observable;
 import rx.Observable.OnSubscribe;
 import rx.Observer;
 import rx.Subscriber;
 import rx.exceptions.CompositeException;
 import rx.exceptions.TestException;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+
+import static org.junit.Assert.*;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyInt;
+import static org.mockito.Mockito.*;
+
 public class OperatorMergeDelayErrorTest {
 
     @Mock
@@ -52,6 +46,36 @@ public void before() {
         MockitoAnnotations.initMocks(this);
     }
 
+    @Test(timeout=1000L)
+    public void testSynchronousError() {
+        final Observable<Observable<String>> o1 = Observable.error(new RuntimeException("unit test"));
+
+        final CountDownLatch latch = new CountDownLatch(1);
+        Observable.mergeDelayError(o1).subscribe(new Subscriber<String>() {
+            @Override
+            public void onCompleted() {
+                fail("Expected onError path");
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                latch.countDown();
+            }
+
+            @Override
+            public void onNext(String s) {
+                fail("Expected onError path");
+            }
+        });
+
+        try {
+            latch.await();
+        } catch (InterruptedException ex) {
+            fail("interrupted");
+        }
+    }
+
+
     @Test
     public void testErrorDelayed1() {
         final Observable<String> o1 = Observable.create(new TestErrorObservable("four", null, "six")); // we expect to lose "six" from the source (and it should never be sent by the source since onError was called