Fixed incorrect error merging.

Author: akarnokd

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

@@ -397,11 +397,13 @@ public Boolean call(InnerSubscriber<T> s) {
 
         @Override
         public void onError(Throwable e) {
-            completed = true;
-            innerError(e);
+            if (!completed) {
+                completed = true;
+                innerError(e, true);
+            }
         }
 
-        private void innerError(Throwable e) {
+        private void innerError(Throwable e, boolean parent) {
             if (delayErrors) {
                 synchronized (this) {
                     if (exceptions == null) {
@@ -411,7 +413,9 @@ private void innerError(Throwable e) {
                 exceptions.add(e);
                 boolean sendOnComplete = false;
                 synchronized (this) {
-                    wip--;
+                    if (!parent) {
+                        wip--;
+                    }
                     if ((wip == 0 && completed) || (wip < 0)) {
                         sendOnComplete = true;
                     }
@@ -545,7 +549,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.innerError(e);
+                parentSubscriber.innerError(e, false);
             }
         }
 

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

@@ -496,26 +496,32 @@ public void testErrorInParentObservable() {
 
     @Test
     public void testErrorInParentObservableDelayed() throws Exception {
-        final TestASynchronous1sDelayedObservable o1 = new TestASynchronous1sDelayedObservable();
-        final TestASynchronous1sDelayedObservable o2 = new TestASynchronous1sDelayedObservable();
-        Observable<Observable<String>> parentObservable = Observable.create(new Observable.OnSubscribe<Observable<String>>() {
-            @Override
-            public void call(Subscriber<? super Observable<String>> op) {
-                op.onNext(Observable.create(o1));
-                op.onNext(Observable.create(o2));
-                op.onError(new NullPointerException("throwing exception in parent"));
-            }
-        });
-
-        TestSubscriber<String> ts = new TestSubscriber<String>(stringObserver);
-        Observable<String> m = Observable.mergeDelayError(parentObservable);
-        m.subscribe(ts);
-        ts.awaitTerminalEvent(2000, TimeUnit.MILLISECONDS);
-        ts.assertTerminalEvent();
-
-        verify(stringObserver, times(2)).onNext("hello");
-        verify(stringObserver, times(1)).onError(any(NullPointerException.class));
-        verify(stringObserver, never()).onCompleted();
+        for (int i = 0; i < 50; i++) {
+            final TestASynchronous1sDelayedObservable o1 = new TestASynchronous1sDelayedObservable();
+            final TestASynchronous1sDelayedObservable o2 = new TestASynchronous1sDelayedObservable();
+            Observable<Observable<String>> parentObservable = Observable.create(new Observable.OnSubscribe<Observable<String>>() {
+                @Override
+                public void call(Subscriber<? super Observable<String>> op) {
+                    op.onNext(Observable.create(o1));
+                    op.onNext(Observable.create(o2));
+                    op.onError(new NullPointerException("throwing exception in parent"));
+                }
+            });
+    
+            @SuppressWarnings("unchecked")
+            Observer<String> stringObserver = mock(Observer.class);
+            
+            TestSubscriber<String> ts = new TestSubscriber<String>(stringObserver);
+            Observable<String> m = Observable.mergeDelayError(parentObservable);
+            m.subscribe(ts);
+            System.out.println("testErrorInParentObservableDelayed | " + i);
+            ts.awaitTerminalEvent(2000, TimeUnit.MILLISECONDS);
+            ts.assertTerminalEvent();
+    
+            verify(stringObserver, times(2)).onNext("hello");
+            verify(stringObserver, times(1)).onError(any(NullPointerException.class));
+            verify(stringObserver, never()).onCompleted();
+        }
     }
 
     private static class TestASynchronous1sDelayedObservable implements Observable.OnSubscribe<String> {