2.x: Fix Completable.andThen(Completable) not running on observeOn scheduler.

src/main/java/io/reactivex/Completable.java

@@ -1171,7 +1171,8 @@ public final <T> Maybe<T> andThen(MaybeSource<T> next) {
     @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Completable andThen(CompletableSource next) {
-        return concatWith(next);
+        ObjectHelper.requireNonNull(next, "next is null");
+        return RxJavaPlugins.onAssembly(new CompletableAndThenCompletable(this, next));
     }
 
     /**
@@ -1357,7 +1358,7 @@ public final Completable compose(CompletableTransformer transformer) {
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Completable concatWith(CompletableSource other) {
         ObjectHelper.requireNonNull(other, "other is null");
-        return concatArray(this, other);
+        return RxJavaPlugins.onAssembly(new CompletableAndThenCompletable(this, other));
     }
 
     /**

src/main/java/io/reactivex/internal/operators/completable/CompletableAndThenCompletable.java

@@ -0,0 +1,107 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.internal.operators.completable;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.reactivex.*;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.disposables.DisposableHelper;
+
+public final class CompletableAndThenCompletable extends Completable {
+
+    final CompletableSource source;
+
+    final CompletableSource next;
+
+    public CompletableAndThenCompletable(CompletableSource source, CompletableSource next) {
+        this.source = source;
+        this.next = next;
+    }
+
+    @Override
+    protected void subscribeActual(CompletableObserver observer) {
+        source.subscribe(new SourceObserver(observer, next));
+    }
+
+    static final class SourceObserver
+            extends AtomicReference<Disposable>
+            implements CompletableObserver, Disposable {
+
+        private static final long serialVersionUID = -4101678820158072998L;
+
+        final CompletableObserver actualObserver;
+
+        final CompletableSource next;
+
+        SourceObserver(CompletableObserver actualObserver, CompletableSource next) {
+            this.actualObserver = actualObserver;
+            this.next = next;
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            if (DisposableHelper.setOnce(this, d)) {
+                actualObserver.onSubscribe(this);
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            actualObserver.onError(e);
+        }
+
+        @Override
+        public void onComplete() {
+            next.subscribe(new NextObserver(this, actualObserver));
+        }
+
+        @Override
+        public void dispose() {
+            DisposableHelper.dispose(this);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return DisposableHelper.isDisposed(get());
+        }
+    }
+
+    static final class NextObserver implements CompletableObserver {
+
+        final AtomicReference<Disposable> parent;
+
+        final CompletableObserver downstream;
+
+        public NextObserver(AtomicReference<Disposable> parent, CompletableObserver downstream) {
+            this.parent = parent;
+            this.downstream = downstream;
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            DisposableHelper.replace(parent, d);
+        }
+
+        @Override
+        public void onComplete() {
+            downstream.onComplete();
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            downstream.onError(e);
+        }
+    }
+}

src/test/java/io/reactivex/internal/operators/completable/CompletableAndThenCompletableabTest.java

@@ -0,0 +1,185 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.internal.operators.completable;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import io.reactivex.*;
+import io.reactivex.exceptions.TestException;
+import io.reactivex.functions.Action;
+import io.reactivex.observers.TestObserver;
+import io.reactivex.schedulers.Schedulers;
+
+import static org.junit.Assert.*;
+
+public class CompletableAndThenCompletableabTest {
+    @Test(expected = NullPointerException.class)
+    public void andThenCompletableCompleteNull() {
+        Completable.complete()
+                .andThen((Completable) null);
+    }
+
+    @Test
+    public void andThenCompletableCompleteComplete() {
+        Completable.complete()
+                .andThen(Completable.complete())
+                .test()
+                .assertComplete();
+    }
+
+    @Test
+    public void andThenCompletableCompleteError() {
+        Completable.complete()
+                .andThen(Completable.error(new TestException("test")))
+                .test()
+                .assertNotComplete()
+                .assertNoValues()
+                .assertError(TestException.class)
+                .assertErrorMessage("test");
+    }
+
+    @Test
+    public void andThenCompletableCompleteNever() {
+        Completable.complete()
+                .andThen(Completable.never())
+                .test()
+                .assertNoValues()
+                .assertNoErrors()
+                .assertNotComplete();
+    }
+
+    @Test
+    public void andThenCompletableErrorComplete() {
+        Completable.error(new TestException("bla"))
+                .andThen(Completable.complete())
+                .test()
+                .assertNotComplete()
+                .assertNoValues()
+                .assertError(TestException.class)
+                .assertErrorMessage("bla");
+    }
+
+    @Test
+    public void andThenCompletableErrorNever() {
+        Completable.error(new TestException("bla"))
+                .andThen(Completable.never())
+                .test()
+                .assertNotComplete()
+                .assertNoValues()
+                .assertError(TestException.class)
+                .assertErrorMessage("bla");
+    }
+
+    @Test
+    public void andThenCompletableErrorError() {
+        Completable.error(new TestException("error1"))
+                .andThen(Completable.error(new TestException("error2")))
+                .test()
+                .assertNotComplete()
+                .assertNoValues()
+                .assertError(TestException.class)
+                .assertErrorMessage("error1");
+    }
+
+    @Test
+    public void andThenCanceled() {
+        final AtomicInteger completableRunCount = new AtomicInteger();
+        Completable.fromRunnable(new Runnable() {
+            @Override
+            public void run() {
+                completableRunCount.incrementAndGet();
+            }
+        })
+                .andThen(Completable.complete())
+                .test(true)
+                .assertEmpty();
+        assertEquals(1, completableRunCount.get());
+    }
+
+    @Test
+    public void andThenFirstCancels() {
+        final TestObserver<Void> to = new TestObserver<Void>();
+        Completable.fromRunnable(new Runnable() {
+            @Override
+            public void run() {
+                to.cancel();
+            }
+        })
+                .andThen(Completable.complete())
+                .subscribe(to);
+        to
+                .assertNotComplete()
+                .assertNoErrors();
+    }
+
+    @Test
+    public void andThenSecondCancels() {
+        final TestObserver<Void> to = new TestObserver<Void>();
+        Completable.complete()
+                .andThen(Completable.fromRunnable(new Runnable() {
+                    @Override
+                    public void run() {
+                        to.cancel();
+                    }
+                }))
+                .subscribe(to);
+        to
+                .assertNotComplete()
+                .assertNoErrors();
+    }
+
+    @Test
+    public void andThenDisposed() {
+        TestHelper.checkDisposed(Completable.complete()
+                .andThen(Completable.complete()));
+    }
+
+    @Test
+    public void andThenNoInterrupt() throws InterruptedException {
+        for (int k = 0; k < 100; k++) {
+            final int count = 10;
+            final CountDownLatch latch = new CountDownLatch(count);
+            final boolean[] interrupted = {false};
+
+            for (int i = 0; i < count; i++) {
+                Completable.complete()
+                        .subscribeOn(Schedulers.io())
+                        .observeOn(Schedulers.io())
+                        .andThen(Completable.fromAction(new Action() {
+                            @Override
+                            public void run() throws Exception {
+                                try {
+                                    Thread.sleep(30);
+                                } catch (InterruptedException e) {
+                                    System.out.println("Interrupted! " + Thread.currentThread());
+                                    interrupted[0] = true;
+                                }
+                            }
+                        }))
+                        .subscribe(new Action() {
+                            @Override
+                            public void run() throws Exception {
+                                latch.countDown();
+                            }
+                        });
+            }
+
+            latch.await();
+            assertFalse("The second Completable was interrupted!", interrupted[0]);
+        }
+    }
+}

src/test/java/io/reactivex/internal/operators/completable/CompletableAndThenTest.java

@@ -13,15 +13,9 @@
 
 package io.reactivex.internal.operators.completable;
 
-import io.reactivex.Completable;
-import io.reactivex.Maybe;
-import io.reactivex.functions.Action;
-import io.reactivex.schedulers.Schedulers;
-
-import java.util.concurrent.CountDownLatch;
-
 import org.junit.Test;
-import static org.junit.Assert.*;
+
+import io.reactivex.*;
 
 public class CompletableAndThenTest {
     @Test(expected = NullPointerException.class)
@@ -69,39 +63,4 @@ public void andThenMaybeError() {
             .assertError(RuntimeException.class)
             .assertErrorMessage("bla");
     }
-
-    @Test
-    public void andThenNoInterrupt() throws InterruptedException {
-        for (int k = 0; k < 100; k++) {
-            final int count = 10;
-            final CountDownLatch latch = new CountDownLatch(count);
-            final boolean[] interrupted = { false };
-
-            for (int i = 0; i < count; i++) {
-                Completable.complete()
-                .subscribeOn(Schedulers.io())
-                .observeOn(Schedulers.io())
-                .andThen(Completable.fromAction(new Action() {
-                    @Override
-                    public void run() throws Exception {
-                        try {
-                            Thread.sleep(30);
-                        } catch (InterruptedException e) {
-                            System.out.println("Interrupted! " + Thread.currentThread());
-                            interrupted[0] = true;
-                        }
-                    }
-                }))
-                .subscribe(new Action() {
-                    @Override
-                    public void run() throws Exception {
-                        latch.countDown();
-                    }
-                });
-            }
-
-            latch.await();
-            assertFalse("The second Completable was interrupted!", interrupted[0]);
-        }
-    }
 }