2.: Fix flatMapX over-cancellation in case of an inner error (#4686)

Author: akarnokd

build.gradle

@@ -115,6 +115,13 @@ jacoco {
     toolVersion = '0.7.7.201606060606' // See http://www.eclemma.org/jacoco/.
 }
 
+task GCandMem(dependsOn: 'check') << {
+    System.gc()
+    Thread.sleep(200)
+    print("Memory usage: ")
+    println(java.lang.management.ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed() / 1024.0 / 1024.0)
+}
+
 jacocoTestReport {
     reports {
         xml.enabled = true
@@ -129,6 +136,8 @@ jacocoTestReport {
     }
 }
 
+jacocoTestReport.dependsOn GCandMem
+
 build.dependsOn jacocoTestReport
 
 // pmd {

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

@@ -228,7 +228,8 @@ void innerError(InnerObserver inner, Throwable e) {
             set.delete(inner);
             if (errors.addThrowable(e)) {
                 if (!delayErrors) {
-                    cancel();
+                    s.cancel();
+                    set.dispose();
                 }
                 active.decrementAndGet();
                 drain();

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

@@ -228,7 +228,8 @@ void innerError(InnerObserver inner, Throwable e) {
             set.delete(inner);
             if (errors.addThrowable(e)) {
                 if (!delayErrors) {
-                    cancel();
+                    s.cancel();
+                    set.dispose();
                 }
                 active.decrementAndGet();
                 drain();

src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapMaybe.java

@@ -195,7 +195,8 @@ void innerError(InnerObserver inner, Throwable e) {
             set.delete(inner);
             if (errors.addThrowable(e)) {
                 if (!delayErrors) {
-                    dispose();
+                    d.dispose();
+                    set.dispose();
                 }
                 active.decrementAndGet();
                 drain();

src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapSingle.java

@@ -195,7 +195,8 @@ void innerError(InnerObserver inner, Throwable e) {
             set.delete(inner);
             if (errors.addThrowable(e)) {
                 if (!delayErrors) {
-                    dispose();
+                    d.dispose();
+                    set.dispose();
                 }
                 active.decrementAndGet();
                 drain();

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

@@ -16,7 +16,7 @@
 import static org.junit.Assert.*;
 
 import java.util.*;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 
 import org.junit.Test;
 
@@ -253,4 +253,22 @@ public MaybeSource<Integer> apply(Integer v) throws Exception {
         .test()
         .assertResult(1, 2);
     }
+
+    @Test
+    public void middleError() {
+        Flowable.fromArray(new String[]{"1","a","2"}).flatMapMaybe(new Function<String, MaybeSource<Integer>>() {
+            @Override
+            public MaybeSource<Integer> apply(final String s) throws NumberFormatException {
+                //return Single.just(Integer.valueOf(s)); //This works
+                return Maybe.fromCallable(new Callable<Integer>() {
+                    @Override
+                    public Integer call() throws NumberFormatException {
+                        return Integer.valueOf(s);
+                    }
+                });
+            }
+        })
+        .test()
+        .assertFailure(NumberFormatException.class, 1);
+    }
 }

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

@@ -16,7 +16,7 @@
 import static org.junit.Assert.*;
 
 import java.util.*;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 
 import org.junit.Test;
 
@@ -240,4 +240,22 @@ public SingleSource<Integer> apply(Integer v) throws Exception {
         .test()
         .assertResult(1, 2);
     }
+
+    @Test
+    public void middleError() {
+        Flowable.fromArray(new String[]{"1","a","2"}).flatMapSingle(new Function<String, SingleSource<Integer>>() {
+            @Override
+            public SingleSource<Integer> apply(final String s) throws NumberFormatException {
+                //return Single.just(Integer.valueOf(s)); //This works
+                return Single.fromCallable(new Callable<Integer>() {
+                    @Override
+                    public Integer call() throws NumberFormatException {
+                        return Integer.valueOf(s);
+                    }
+                });
+            }
+        })
+        .test()
+        .assertFailure(NumberFormatException.class, 1);
+    }
 }

src/test/java/io/reactivex/internal/operators/observable/ObservableFlatMapMaybeTest.java

@@ -16,7 +16,7 @@
 import static org.junit.Assert.*;
 
 import java.util.*;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 
 import org.junit.Test;
 
@@ -181,4 +181,22 @@ public MaybeSource<Integer> apply(Integer v) throws Exception {
         .test()
         .assertResult(1, 2);
     }
+
+    @Test
+    public void middleError() {
+        Observable.fromArray(new String[]{"1","a","2"}).flatMapMaybe(new Function<String, MaybeSource<Integer>>() {
+            @Override
+            public MaybeSource<Integer> apply(final String s) throws NumberFormatException {
+                //return Single.just(Integer.valueOf(s)); //This works
+                return Maybe.fromCallable(new Callable<Integer>() {
+                    @Override
+                    public Integer call() throws NumberFormatException {
+                        return Integer.valueOf(s);
+                    }
+                });
+            }
+        })
+        .test()
+        .assertFailure(NumberFormatException.class, 1);
+    }
 }

src/test/java/io/reactivex/internal/operators/observable/ObservableFlatMapSingleTest.java

@@ -16,7 +16,7 @@
 import static org.junit.Assert.*;
 
 import java.util.*;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 
 import org.junit.Test;
 
@@ -168,4 +168,22 @@ public SingleSource<Integer> apply(Integer v) throws Exception {
         .test()
         .assertResult(1, 2);
     }
+
+    @Test
+    public void middleError() {
+        Observable.fromArray(new String[]{"1","a","2"}).flatMapSingle(new Function<String, SingleSource<Integer>>() {
+            @Override
+            public SingleSource<Integer> apply(final String s) throws NumberFormatException {
+                //return Single.just(Integer.valueOf(s)); //This works
+                return Single.fromCallable(new Callable<Integer>() {
+                    @Override
+                    public Integer call() throws NumberFormatException {
+                        return Integer.valueOf(s);
+                    }
+                });
+            }
+        })
+        .test()
+        .assertFailure(NumberFormatException.class, 1);
+    }
 }