Fix mergeDelayError Handling of Error in Parent Observable

Fixes #313

Author: benjchristensen

src/main/java/rx/internal/operators/OperatorMerge.java

@@ -397,6 +397,11 @@ public Boolean call(InnerSubscriber<T> s) {
 
         @Override
         public void onError(Throwable e) {
+            completed = true;
+            innerError(e);
+        }
+        
+        private void innerError(Throwable e) {
             if (delayErrors) {
                 synchronized (this) {
                     if (exceptions == null) {
@@ -540,7 +545,7 @@ public void onNext(T t) {
         public void onError(Throwable e) {
             // it doesn't go through queues, it immediately onErrors and tears everything down
             if (ONCE_TERMINATED.compareAndSet(this, 0, 1)) {
-                parentSubscriber.onError(e);
+                parentSubscriber.innerError(e);
             }
         }
 
@@ -753,4 +758,4 @@ private int drainQueue() {
             }
         }
     }
-}
+}

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

@@ -30,6 +30,7 @@
 import rx.observers.TestSubscriber;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -479,6 +480,20 @@ public void onCompleted() {
         verify(o, never()).onCompleted();
     }
 
+    @Test
+    public void testErrorInParentObservable() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        Observable.mergeDelayError(
+                Observable.just(Observable.just(1), Observable.just(2))
+                        .startWith(Observable.<Integer> error(new RuntimeException()))
+                ).subscribe(ts);
+        ts.awaitTerminalEvent();
+        ts.assertTerminalEvent();
+        ts.assertReceivedOnNext(Arrays.asList(1, 2));
+        assertEquals(1, ts.getOnErrorEvents().size());
+
+    }
+
     @Test
     public void testErrorInParentObservableDelayed() throws Exception {
         final TestASynchronous1sDelayedObservable o1 = new TestASynchronous1sDelayedObservable();