From d2c303fa36321b138d0d1f379fe79c79cf8abfee Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?D=C3=A1vid=20Karnok?= <akarnokd@gmail.com>
Date: Tue, 12 Feb 2019 21:30:18 +0100
Subject: [PATCH] 2.x: Fix concatEager to dispose sources & clean up properly.

---
 .../flowable/FlowableConcatMapEager.java      |  7 +++-
 .../observable/ObservableConcatMapEager.java  | 15 +++++++--
 .../flowable/FlowableConcatMapEagerTest.java  | 33 +++++++++++++++++++
 .../ObservableConcatMapEagerTest.java         | 33 +++++++++++++++++++
 4 files changed, 85 insertions(+), 3 deletions(-)

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatMapEager.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatMapEager.java
index 87ee235704..8acad8ac69 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatMapEager.java
+++ b/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();
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableConcatMapEager.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableConcatMapEager.java
index cb15b10f65..7028fdcf62 100644
--- a/src/main/java/io/reactivex/internal/operators/observable/ObservableConcatMapEager.java
+++ b/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);
             }
         }
 
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatMapEagerTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatMapEagerTest.java
index 1e34d2937b..11f12fb4a3 100644
--- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatMapEagerTest.java
+++ b/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());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableConcatMapEagerTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableConcatMapEagerTest.java
index 3959fab45e..17e418bab7 100644
--- a/src/test/java/io/reactivex/internal/operators/observable/ObservableConcatMapEagerTest.java
+++ b/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());
+    }
 }