From 1b70eb0afc2123658ddd5bb9fecbdfdf02aeabaf Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?D=C3=A1vid=20Karnok?= <akarnokd@gmail.com>
Date: Thu, 7 Apr 2016 17:37:31 +0200
Subject: [PATCH 1/2] 1.x: fix concatMap scalar source behavior

---
 .../internal/operators/OnSubscribeConcatMap.java   |  4 +++-
 .../rx/internal/operators/OperatorConcatTest.java  | 14 ++++++++++++++
 2 files changed, 17 insertions(+), 1 deletion(-)

diff --git a/src/main/java/rx/internal/operators/OnSubscribeConcatMap.java b/src/main/java/rx/internal/operators/OnSubscribeConcatMap.java
index 001058763b..6f0e8435b4 100644
--- a/src/main/java/rx/internal/operators/OnSubscribeConcatMap.java
+++ b/src/main/java/rx/internal/operators/OnSubscribeConcatMap.java
@@ -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);
@@ -352,7 +354,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);
diff --git a/src/test/java/rx/internal/operators/OperatorConcatTest.java b/src/test/java/rx/internal/operators/OperatorConcatTest.java
index 65aa1f5307..8a39b16f9f 100644
--- a/src/test/java/rx/internal/operators/OperatorConcatTest.java
+++ b/src/test/java/rx/internal/operators/OperatorConcatTest.java
@@ -850,4 +850,18 @@ 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();
+    }
 }
\ No newline at end of file

From eff6bb03eb333921bcb132159a92f2cf6717681d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?D=C3=A1vid=20Karnok?= <akarnokd@gmail.com>
Date: Thu, 7 Apr 2016 17:51:00 +0200
Subject: [PATCH 2/2] More tests and fix empty() case as well

---
 .../operators/OnSubscribeConcatMap.java       | 12 +++--
 .../operators/OperatorConcatTest.java         | 46 +++++++++++++++++++
 2 files changed, 55 insertions(+), 3 deletions(-)

diff --git a/src/main/java/rx/internal/operators/OnSubscribeConcatMap.java b/src/main/java/rx/internal/operators/OnSubscribeConcatMap.java
index 6f0e8435b4..c2799df758 100644
--- a/src/main/java/rx/internal/operators/OnSubscribeConcatMap.java
+++ b/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;
                 }
@@ -288,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) {
diff --git a/src/test/java/rx/internal/operators/OperatorConcatTest.java b/src/test/java/rx/internal/operators/OperatorConcatTest.java
index 8a39b16f9f..a824374659 100644
--- a/src/test/java/rx/internal/operators/OperatorConcatTest.java
+++ b/src/test/java/rx/internal/operators/OperatorConcatTest.java
@@ -864,4 +864,50 @@ public void scalarAndRangeBackpressured() {
         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();
+    }
+
 }
\ No newline at end of file