Fix Merge: backpressure + scalarValueQueue don't play nicely

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

@@ -72,9 +72,6 @@ public class OperatorMerge<T> implements Operator<T, Observable<? extends T>> {
      * I'd love to have contributions that improve this class, but keep in mind the performance and GC pressure.
      * The benchmarks I use are in the JMH OperatorMergePerf class. GC memory pressure is tested using Java Flight Recorder
      * to track object allocation.
-     * 
-     * TODO There is still a known concurrency bug somewhere either in this class, in SubscriptionIndexedRingBuffer or their relationship.
-     * See https://github.com/ReactiveX/RxJava/issues/1420 for more information on this.
      */
 
     public OperatorMerge() {
@@ -426,7 +423,7 @@ public void onCompleted() {
             boolean c = false;
             synchronized (this) {
                 completed = true;
-                if (wip == 0) {
+                if (wip == 0 && (scalarValueQueue == null || scalarValueQueue.isEmpty())) {
                     c = true;
                 }
             }
@@ -496,7 +493,17 @@ public void request(long n) {
                 requested = Long.MAX_VALUE;
             } else {
                 REQUESTED.getAndAdd(this, n);
-                ms.drainQueuesIfNeeded();
+                if (ms.drainQueuesIfNeeded()) {
+                    boolean sendComplete = false;
+                    synchronized (this) {
+                        if (ms.wip == 0 && ms.scalarValueQueue != null && ms.scalarValueQueue.isEmpty()) {
+                            sendComplete = true;
+                        }
+                    }
+                    if (sendComplete) {
+                        ms.drainAndComplete();
+                    }
+                }
             }
         }
 

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

@@ -15,6 +15,7 @@
  */
 package rx.internal.operators;
 
+import static java.util.Arrays.asList;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
@@ -39,13 +40,9 @@
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 
-import rx.Observable;
+import rx.*;
 import rx.Observable.OnSubscribe;
-import rx.Observer;
-import rx.Scheduler;
 import rx.Scheduler.Worker;
-import rx.Subscriber;
-import rx.Subscription;
 import rx.functions.Action0;
 import rx.functions.Action1;
 import rx.functions.Func1;
@@ -1005,4 +1002,107 @@ public void call(Subscriber<? super Integer> s) {
         assertEquals(10000, ts.getOnNextEvents().size());
     }
 
+    @Test
+    public void shouldCompleteAfterApplyingBackpressure_NormalPath() {
+        Observable<Integer> source = Observable.mergeDelayError(Observable.just(Observable.range(1, 2)));
+        TestSubscriber<Integer> subscriber = new TestSubscriber<Integer>();
+        subscriber.requestMore(0);
+        source.subscribe(subscriber);
+        subscriber.requestMore(3); // 1, 2, <complete> - with requestMore(2) we get the 1 and 2 but not the <complete>
+        subscriber.assertReceivedOnNext(asList(1, 2));
+        subscriber.assertTerminalEvent();
+    }
+
+    @Test
+    public void shouldCompleteAfterApplyingBackpressure_FastPath() {
+        Observable<Integer> source = Observable.mergeDelayError(Observable.just(Observable.just(1)));
+        TestSubscriber<Integer> subscriber = new TestSubscriber<Integer>();
+        subscriber.requestMore(0);
+        source.subscribe(subscriber);
+        subscriber.requestMore(2); // 1, <complete> - should work as per .._NormalPath above
+        subscriber.assertReceivedOnNext(asList(1));
+        subscriber.assertTerminalEvent();
+    }
+
+    @Test
+    public void shouldNotCompleteIfThereArePendingScalarSynchronousEmissionsWhenTheLastInnerSubscriberCompletes() {
+        TestScheduler scheduler = Schedulers.test();
+        Observable<Long> source = Observable.mergeDelayError(Observable.just(1L), Observable.timer(1, TimeUnit.SECONDS, scheduler).skip(1));
+        TestSubscriber<Long> subscriber = new TestSubscriber<Long>();
+        subscriber.requestMore(0);
+        source.subscribe(subscriber);
+        scheduler.advanceTimeBy(1, TimeUnit.SECONDS);
+        subscriber.assertReceivedOnNext(Collections.<Long>emptyList());
+        assertEquals(Collections.<Notification<Long>>emptyList(), subscriber.getOnCompletedEvents());
+        subscriber.requestMore(1);
+        subscriber.assertReceivedOnNext(asList(1L));
+        assertEquals(Collections.<Notification<Long>>emptyList(), subscriber.getOnCompletedEvents());
+        subscriber.requestMore(1);
+        subscriber.assertTerminalEvent();
+    }
+
+    @Test
+    public void delayedErrorsShouldBeEmittedWhenCompleteAfterApplyingBackpressure_NormalPath() {
+        Throwable exception = new Throwable();
+        Observable<Integer> source = Observable.mergeDelayError(Observable.range(1, 2), Observable.<Integer>error(exception));
+        TestSubscriber<Integer> subscriber = new TestSubscriber<Integer>();
+        subscriber.requestMore(0);
+        source.subscribe(subscriber);
+        subscriber.requestMore(3); // 1, 2, <error>
+        subscriber.assertReceivedOnNext(asList(1, 2));
+        subscriber.assertTerminalEvent();
+        assertEquals(asList(exception), subscriber.getOnErrorEvents());
+    }
+
+    @Test
+    public void delayedErrorsShouldBeEmittedWhenCompleteAfterApplyingBackpressure_FastPath() {
+        Throwable exception = new Throwable();
+        Observable<Integer> source = Observable.mergeDelayError(Observable.just(1), Observable.<Integer>error(exception));
+        TestSubscriber<Integer> subscriber = new TestSubscriber<Integer>();
+        subscriber.requestMore(0);
+        source.subscribe(subscriber);
+        subscriber.requestMore(2); // 1, <error>
+        subscriber.assertReceivedOnNext(asList(1));
+        subscriber.assertTerminalEvent();
+        assertEquals(asList(exception), subscriber.getOnErrorEvents());
+    }
+
+    @Test
+    public void shouldNotCompleteWhileThereAreStillScalarSynchronousEmissionsInTheQueue() {
+        Observable<Integer> source = Observable.merge(Observable.just(1), Observable.just(2));
+        TestSubscriber<Integer> subscriber = new TestSubscriber<Integer>();
+        subscriber.requestMore(1);
+        source.subscribe(subscriber);
+        subscriber.assertReceivedOnNext(asList(1));
+        subscriber.requestMore(1);
+        subscriber.assertReceivedOnNext(asList(1, 2));
+    }
+
+    @Test
+    public void shouldNotReceivedDelayedErrorWhileThereAreStillScalarSynchronousEmissionsInTheQueue() {
+        Throwable exception = new Throwable();
+        Observable<Integer> source = Observable.mergeDelayError(Observable.just(1), Observable.just(2), Observable.<Integer>error(exception));
+        TestSubscriber<Integer> subscriber = new TestSubscriber<Integer>();
+        subscriber.requestMore(1);
+        source.subscribe(subscriber);
+        subscriber.assertReceivedOnNext(asList(1));
+        assertEquals(Collections.<Throwable>emptyList(), subscriber.getOnErrorEvents());
+        subscriber.requestMore(1);
+        subscriber.assertReceivedOnNext(asList(1, 2));
+        assertEquals(asList(exception), subscriber.getOnErrorEvents());
+    }
+
+    @Test
+    public void shouldNotReceivedDelayedErrorWhileThereAreStillNormalEmissionsInTheQueue() {
+        Throwable exception = new Throwable();
+        Observable<Integer> source = Observable.mergeDelayError(Observable.range(1, 2), Observable.range(3, 2), Observable.<Integer>error(exception));
+        TestSubscriber<Integer> subscriber = new TestSubscriber<Integer>();
+        subscriber.requestMore(3);
+        source.subscribe(subscriber);
+        subscriber.assertReceivedOnNext(asList(1, 2, 3));
+        assertEquals(Collections.<Throwable>emptyList(), subscriber.getOnErrorEvents());
+        subscriber.requestMore(2);
+        subscriber.assertReceivedOnNext(asList(1, 2, 3, 4));
+        assertEquals(asList(exception), subscriber.getOnErrorEvents());
+    }
 }