1.x: fix Completable.concat to use replace (don't dispose old) (#5696)

* 1.x: fix Completable.concat to use replace (don't dispose old)

* Remove original issue comments

Author: akarnokd

src/main/java/rx/internal/operators/CompletableOnSubscribeConcatArray.java

@@ -20,7 +20,7 @@
 
 import rx.*;
 import rx.Completable.OnSubscribe;
-import rx.subscriptions.SerialSubscription;
+import rx.internal.subscriptions.SequentialSubscription;
 
 public final class CompletableOnSubscribeConcatArray implements OnSubscribe {
     final Completable[] sources;
@@ -45,17 +45,17 @@ static final class ConcatInnerSubscriber extends AtomicInteger implements Comple
 
         int index;
 
-        final SerialSubscription sd;
+        final SequentialSubscription sd;
 
         public ConcatInnerSubscriber(CompletableSubscriber actual, Completable[] sources) {
             this.actual = actual;
             this.sources = sources;
-            this.sd = new SerialSubscription();
+            this.sd = new SequentialSubscription();
         }
 
         @Override
         public void onSubscribe(Subscription d) {
-            sd.set(d);
+            sd.replace(d);
         }
 
         @Override

src/main/java/rx/internal/operators/CompletableOnSubscribeConcatIterable.java

@@ -21,7 +21,8 @@
 
 import rx.*;
 import rx.Completable.OnSubscribe;
-import rx.subscriptions.*;
+import rx.internal.subscriptions.SequentialSubscription;
+import rx.subscriptions.Subscriptions;
 
 public final class CompletableOnSubscribeConcatIterable implements OnSubscribe {
     final Iterable<? extends Completable> sources;
@@ -61,17 +62,17 @@ static final class ConcatInnerSubscriber extends AtomicInteger implements Comple
         final CompletableSubscriber actual;
         final Iterator<? extends Completable> sources;
 
-        final SerialSubscription sd;
+        final SequentialSubscription sd;
 
         public ConcatInnerSubscriber(CompletableSubscriber actual, Iterator<? extends Completable> sources) {
             this.actual = actual;
             this.sources = sources;
-            this.sd = new SerialSubscription();
+            this.sd = new SequentialSubscription();
         }
 
         @Override
         public void onSubscribe(Subscription d) {
-            sd.set(d);
+            sd.replace(d);
         }
 
         @Override

src/test/java/rx/internal/operators/CompletableConcatTest.java

@@ -16,7 +16,11 @@
 
 package rx.internal.operators;
 
-import java.util.concurrent.TimeUnit;
+import static org.junit.Assert.assertFalse;
+
+import java.util.Arrays;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.*;
 
@@ -58,4 +62,77 @@ public void call() {
 
         Assert.assertEquals(5, calls[0]);
     }
+
+    @Test
+    public void andThenNoInterrupt() throws InterruptedException {
+        for (int k = 0; k < 100; k++) {
+            final int count = 10;
+            final CountDownLatch latch = new CountDownLatch(count);
+            final AtomicBoolean interrupted = new AtomicBoolean();
+
+            for (int i = 0; i < count; i++) {
+                Completable.complete()
+                .subscribeOn(Schedulers.io())
+                .observeOn(Schedulers.io())
+                .andThen(Completable.fromAction(new Action0() {
+                    @Override
+                    public void call() {
+                        try {
+                            Thread.sleep(30);
+                        } catch (InterruptedException e) {
+                            System.out.println("Interrupted! " + Thread.currentThread());
+                            interrupted.set(true);
+                        }
+                    }
+                }))
+                .subscribe(new Action0() {
+                    @Override
+                    public void call() {
+                        latch.countDown();
+                    }
+                });
+            }
+
+            latch.await();
+            assertFalse("The second Completable was interrupted!", interrupted.get());
+        }
+    }
+
+    @Test
+    public void noInterrupt() throws InterruptedException {
+        for (int k = 0; k < 100; k++) {
+            final int count = 10;
+            final CountDownLatch latch = new CountDownLatch(count);
+            final AtomicBoolean interrupted = new AtomicBoolean();
+
+            for (int i = 0; i < count; i++) {
+                Completable c0 = Completable.fromAction(new Action0() {
+                    @Override
+                    public void call() {
+                        try {
+                            Thread.sleep(30);
+                        } catch (InterruptedException e) {
+                            System.out.println("Interrupted! " + Thread.currentThread());
+                            interrupted.set(true);
+                        }
+                    }
+                });
+                Completable.concat(Arrays.asList(Completable.complete()
+                    .subscribeOn(Schedulers.io())
+                    .observeOn(Schedulers.io()),
+                    c0)
+                )
+                .subscribe(new Action0() {
+                    @Override
+                    public void call() {
+                        latch.countDown();
+                    }
+                });
+            }
+
+            latch.await();
+            assertFalse("The second Completable was interrupted!", interrupted.get());
+        }
+    }
+
 }