2.x: Fix concatEager to dispose sources & clean up properly. (#6405)

Author: akarnokd

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

@@ -176,7 +176,12 @@ void drainAndCancel() {
         }
 
         void cancelAll() {
-            InnerQueuedSubscriber<R> inner;
+            InnerQueuedSubscriber<R> inner = current;
+            current = null;
+
+            if (inner != null) {
+                inner.cancel();
+            }
 
             while ((inner = subscribers.poll()) != null) {
                 inner.cancel();

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

@@ -162,10 +162,21 @@ public void onComplete() {
 
         @Override
         public void dispose() {
+            if (cancelled) {
+                return;
+            }
             cancelled = true;
+            upstream.dispose();
+
+            drainAndDispose();
+        }
+
+        void drainAndDispose() {
             if (getAndIncrement() == 0) {
-                queue.clear();
-                disposeAll();
+                do {
+                    queue.clear();
+                    disposeAll();
+                } while (decrementAndGet() != 0);
             }
         }
 

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

@@ -1333,4 +1333,37 @@ public void arrayDelayErrorMaxConcurrencyErrorDelayed() {
 
         ts.assertFailure(TestException.class, 1, 2);
     }
+
+    @Test
+    public void cancelActive() {
+        PublishProcessor<Integer> pp1 = PublishProcessor.create();
+        PublishProcessor<Integer> pp2 = PublishProcessor.create();
+
+        TestSubscriber<Integer> ts = Flowable
+                .concatEager(Flowable.just(pp1, pp2))
+                .test();
+
+        assertTrue(pp1.hasSubscribers());
+        assertTrue(pp2.hasSubscribers());
+
+        ts.cancel();
+
+        assertFalse(pp1.hasSubscribers());
+        assertFalse(pp2.hasSubscribers());
+    }
+
+    @Test
+    public void cancelNoInnerYet() {
+        PublishProcessor<Flowable<Integer>> pp1 = PublishProcessor.create();
+
+        TestSubscriber<Integer> ts = Flowable
+                .concatEager(pp1)
+                .test();
+
+        assertTrue(pp1.hasSubscribers());
+
+        ts.cancel();
+
+        assertFalse(pp1.hasSubscribers());
+    }
 }

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

@@ -1141,4 +1141,37 @@ public void arrayDelayErrorMaxConcurrencyErrorDelayed() {
 
         to.assertFailure(TestException.class, 1, 2);
     }
+
+    @Test
+    public void cancelActive() {
+        PublishSubject<Integer> ps1 = PublishSubject.create();
+        PublishSubject<Integer> ps2 = PublishSubject.create();
+
+        TestObserver<Integer> to = Observable
+                .concatEager(Observable.just(ps1, ps2))
+                .test();
+
+        assertTrue(ps1.hasObservers());
+        assertTrue(ps2.hasObservers());
+
+        to.dispose();
+
+        assertFalse(ps1.hasObservers());
+        assertFalse(ps2.hasObservers());
+    }
+
+    @Test
+    public void cancelNoInnerYet() {
+        PublishSubject<Observable<Integer>> ps1 = PublishSubject.create();
+
+        TestObserver<Integer> to = Observable
+                .concatEager(ps1)
+                .test();
+
+        assertTrue(ps1.hasObservers());
+
+        to.dispose();
+
+        assertFalse(ps1.hasObservers());
+    }
 }