Merge pull request #3840 from akarnokd/ConcatMapFix

1.x: fix concatMap scalar/empty source behavior

Author: zsxwing

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

@@ -220,7 +220,7 @@ void drain() {
 
             final int delayErrorMode = this.delayErrorMode;
 
-            do {
+            for (;;) {
                 if (actual.isUnsubscribed()) {
                     return;
                 }
@@ -273,6 +273,8 @@ void drain() {
                             if (source instanceof ScalarSynchronousObservable) {
                                 ScalarSynchronousObservable<? extends R> scalarSource = (ScalarSynchronousObservable<? extends R>) source;
 
+                                active = true;
+                                
                                 arbiter.setProducer(new ConcatMapInnerScalarProducer<T, R>(scalarSource.get(), this));
                             } else {
                                 ConcatMapInnerSubscriber<T, R> innerSubscriber = new ConcatMapInnerSubscriber<T, R>(this);
@@ -286,11 +288,17 @@ void drain() {
                                     return;
                                 }
                             }
+                            request(1);
+                        } else {
+                            request(1);
+                            continue;
                         }
-                        request(1);
                     }
                 }
-            } while (wip.decrementAndGet() != 0);
+                if (wip.decrementAndGet() == 0) {
+                    break;
+                }
+            }
         }
 
         void drainError(Throwable mapperError) {
@@ -352,7 +360,7 @@ public ConcatMapInnerScalarProducer(R value, ConcatMapSubscriber<T, R> parent) {
 
         @Override
         public void request(long n) {
-            if (!once) {
+            if (!once && n > 0L) {
                 once = true;
                 ConcatMapSubscriber<T, R> p = parent;
                 p.innerNext(value);

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

@@ -850,4 +850,64 @@ public Observable<Integer> call(Integer t) {
         }
     }
 
+    @Test
+    public void scalarAndRangeBackpressured() {
+        TestSubscriber<Integer> ts = TestSubscriber.create(0);
+        
+        Observable.just(1).concatWith(Observable.range(2, 3)).subscribe(ts);
+        
+        ts.assertNoValues();
+        
+        ts.requestMore(5);
+        
+        ts.assertValues(1, 2, 3, 4);
+        ts.assertCompleted();
+        ts.assertNoErrors();
+    }
+    
+    @Test
+    public void scalarAndEmptyBackpressured() {
+        TestSubscriber<Integer> ts = TestSubscriber.create(0);
+        
+        Observable.just(1).concatWith(Observable.<Integer>empty()).subscribe(ts);
+        
+        ts.assertNoValues();
+        
+        ts.requestMore(5);
+        
+        ts.assertValue(1);
+        ts.assertCompleted();
+        ts.assertNoErrors();
+    }
+
+    @Test
+    public void rangeAndEmptyBackpressured() {
+        TestSubscriber<Integer> ts = TestSubscriber.create(0);
+        
+        Observable.range(1, 2).concatWith(Observable.<Integer>empty()).subscribe(ts);
+        
+        ts.assertNoValues();
+        
+        ts.requestMore(5);
+        
+        ts.assertValues(1, 2);
+        ts.assertCompleted();
+        ts.assertNoErrors();
+    }
+
+    @Test
+    public void emptyAndScalarBackpressured() {
+        TestSubscriber<Integer> ts = TestSubscriber.create(0);
+        
+        Observable.<Integer>empty().concatWith(Observable.just(1)).subscribe(ts);
+        
+        ts.assertNoValues();
+        
+        ts.requestMore(5);
+        
+        ts.assertValue(1);
+        ts.assertCompleted();
+        ts.assertNoErrors();
+    }
+
 }