2.x: Add Completable.andThen(MaybeSource) (#4616)

* 2.x: Add Completable.andThen(MaybeSource)

* Add missing import

* Requested changes

Author: vanniktech

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

@@ -26,6 +26,7 @@
 import io.reactivex.internal.operators.completable.*;
 import io.reactivex.internal.operators.flowable.FlowableDelaySubscriptionOther;
 import io.reactivex.internal.operators.maybe.MaybeFromCompletable;
+import io.reactivex.internal.operators.maybe.MaybeDelayWithCompletable;
 import io.reactivex.internal.operators.observable.ObservableDelaySubscriptionOther;
 import io.reactivex.internal.operators.single.SingleDelayWithCompletable;
 import io.reactivex.internal.util.ExceptionHelper;
@@ -799,6 +800,26 @@ public final <T> Single<T> andThen(SingleSource<T> next) {
         return RxJavaPlugins.onAssembly(new SingleDelayWithCompletable<T>(next, this));
     }
 
+    /**
+     * Returns a {@link Maybe} which will subscribe to this Completable and once that is completed then
+     * will subscribe to the {@code next} MaybeSource. An error event from this Completable will be
+     * propagated to the downstream subscriber and will result in skipping the subscription of the
+     * Maybe.
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code andThen} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param <T> the value type of the next MaybeSource
+     * @param next the Maybe to subscribe after this Completable is completed, not null
+     * @return Maybe that composes this Completable and next
+     */
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final <T> Maybe<T> andThen(MaybeSource<T> next) {
+        ObjectHelper.requireNonNull(next, "next is null");
+        return RxJavaPlugins.onAssembly(new MaybeDelayWithCompletable<T>(next, this));
+    }
+
     /**
      * Returns a Completable that first runs this Completable
      * and then the other completable.

src/main/java/io/reactivex/internal/operators/maybe/MaybeDelayWithCompletable.java

@@ -0,0 +1,115 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * 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.maybe;
+
+import io.reactivex.CompletableObserver;
+import io.reactivex.CompletableSource;
+import io.reactivex.Maybe;
+import io.reactivex.MaybeObserver;
+import io.reactivex.MaybeSource;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.disposables.DisposableHelper;
+import java.util.concurrent.atomic.AtomicReference;
+
+public final class MaybeDelayWithCompletable<T> extends Maybe<T> {
+
+    final MaybeSource<T> source;
+
+    final CompletableSource other;
+
+    public MaybeDelayWithCompletable(MaybeSource<T> source, CompletableSource other) {
+        this.source = source;
+        this.other = other;
+    }
+
+    @Override
+    protected void subscribeActual(MaybeObserver<? super T> subscriber) {
+        other.subscribe(new OtherObserver<T>(subscriber, source));
+    }
+
+    static final class OtherObserver<T>
+    extends AtomicReference<Disposable>
+    implements CompletableObserver, Disposable {
+        private static final long serialVersionUID = 703409937383992161L;
+
+        final MaybeObserver<? super T> actual;
+
+        final MaybeSource<T> source;
+
+        OtherObserver(MaybeObserver<? super T> actual, MaybeSource<T> source) {
+            this.actual = actual;
+            this.source = source;
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            if (DisposableHelper.setOnce(this, d)) {
+
+                actual.onSubscribe(this);
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            actual.onError(e);
+        }
+
+        @Override
+        public void onComplete() {
+            source.subscribe(new DelayWithMainObserver<T>(this, actual));
+        }
+
+        @Override
+        public void dispose() {
+            DisposableHelper.dispose(this);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return DisposableHelper.isDisposed(get());
+        }
+    }
+
+    static final class DelayWithMainObserver<T> implements MaybeObserver<T> {
+
+        final AtomicReference<Disposable> parent;
+
+        final MaybeObserver<? super T> actual;
+
+        DelayWithMainObserver(AtomicReference<Disposable> parent, MaybeObserver<? super T> actual) {
+            this.parent = parent;
+            this.actual = actual;
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            DisposableHelper.replace(parent, d);
+        }
+
+        @Override
+        public void onSuccess(T value) {
+            actual.onSuccess(value);
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            actual.onError(e);
+        }
+
+        @Override
+        public void onComplete() {
+            actual.onComplete();
+        }
+    }
+}

src/main/java/io/reactivex/internal/operators/single/SingleDelayWithCompletable.java

@@ -53,7 +53,7 @@ static final class OtherObserver<T>
 
         @Override
         public void onSubscribe(Disposable d) {
-            if (DisposableHelper.set(this, d)) {
+            if (DisposableHelper.setOnce(this, d)) {
 
                 actual.onSubscribe(this);
             }

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

@@ -0,0 +1,66 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * 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 io.reactivex.Completable;
+import io.reactivex.Maybe;
+import org.junit.Test;
+
+public class CompletableAndThenTest {
+    @Test(expected = NullPointerException.class)
+    public void andThenMaybeNull() {
+        Completable.complete()
+            .andThen((Maybe<Object>) null);
+    }
+
+    @Test
+    public void andThenMaybeCompleteValue() {
+        Completable.complete()
+            .andThen(Maybe.just(1))
+            .test()
+            .assertResult(1);
+    }
+
+    @Test
+    public void andThenMaybeCompleteError() {
+        Completable.complete()
+            .andThen(Maybe.error(new RuntimeException("test")))
+            .test()
+            .assertNotComplete()
+            .assertNoValues()
+            .assertError(RuntimeException.class)
+            .assertErrorMessage("test");
+    }
+
+    @Test
+    public void andThenMaybeCompleteEmpty() {
+        Completable.complete()
+            .andThen(Maybe.empty())
+            .test()
+            .assertNoValues()
+            .assertNoErrors()
+            .assertComplete();
+    }
+
+    @Test
+    public void andThenMaybeError() {
+        Completable.error(new RuntimeException("bla"))
+            .andThen(Maybe.empty())
+            .test()
+            .assertNotComplete()
+            .assertNoValues()
+            .assertError(RuntimeException.class)
+            .assertErrorMessage("bla");
+    }
+}