2.x: fix Flowable.flatMapMaybe/Single maxConcurrency not requesting more (#5287)

Author: akarnokd

src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapMaybe.java

@@ -230,6 +230,10 @@ void innerError(InnerObserver inner, Throwable e) {
                 if (!delayErrors) {
                     s.cancel();
                     set.dispose();
+                } else {
+                    if (maxConcurrency != Integer.MAX_VALUE) {
+                        s.request(1);
+                    }
                 }
                 active.decrementAndGet();
                 drain();
@@ -254,12 +258,19 @@ void innerComplete(InnerObserver inner) {
                     }
                     return;
                 }
+
+                if (maxConcurrency != Integer.MAX_VALUE) {
+                    s.request(1);
+                }
                 if (decrementAndGet() == 0) {
                     return;
                 }
                 drainLoop();
             } else {
                 active.decrementAndGet();
+                if (maxConcurrency != Integer.MAX_VALUE) {
+                    s.request(1);
+                }
                 drain();
             }
         }

src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapSingle.java

@@ -230,6 +230,10 @@ void innerError(InnerObserver inner, Throwable e) {
                 if (!delayErrors) {
                     s.cancel();
                     set.dispose();
+                } else {
+                    if (maxConcurrency != Integer.MAX_VALUE) {
+                        s.request(1);
+                    }
                 }
                 active.decrementAndGet();
                 drain();

src/test/java/io/reactivex/internal/operators/flowable/FlowableFlatMapMaybeTest.java

@@ -320,6 +320,36 @@ public MaybeSource<Integer> apply(Integer v) throws Exception {
         .assertResult();
     }
 
+    @Test
+    public void asyncFlattenNoneMaxConcurrency() {
+        Flowable.range(1, 1000)
+        .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
+            @Override
+            public MaybeSource<Integer> apply(Integer v) throws Exception {
+                return Maybe.<Integer>empty().subscribeOn(Schedulers.computation());
+            }
+        }, false, 128)
+        .take(500)
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult();
+    }
+
+    @Test
+    public void asyncFlattenErrorMaxConcurrency() {
+        Flowable.range(1, 1000)
+        .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
+            @Override
+            public MaybeSource<Integer> apply(Integer v) throws Exception {
+                return Maybe.<Integer>error(new TestException()).subscribeOn(Schedulers.computation());
+            }
+        }, true, 128)
+        .take(500)
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertFailure(CompositeException.class);
+    }
+
     @Test
     public void successError() {
         final PublishProcessor<Integer> ps = PublishProcessor.create();

src/test/java/io/reactivex/internal/operators/flowable/FlowableFlatMapSingleTest.java

@@ -490,4 +490,19 @@ public void run() {
             TestHelper.race(r1, r2);
         }
     }
+
+    @Test
+    public void asyncFlattenErrorMaxConcurrency() {
+        Flowable.range(1, 1000)
+        .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
+            @Override
+            public MaybeSource<Integer> apply(Integer v) throws Exception {
+                return Maybe.<Integer>error(new TestException()).subscribeOn(Schedulers.computation());
+            }
+        }, true, 128)
+        .take(500)
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertFailure(CompositeException.class);
+    }
 }