2.x: Add Single.flatMapMaybe

src/main/java/io/reactivex/Single.java

@@ -1817,7 +1817,7 @@ public final Maybe<T> filter(Predicate<? super T> predicate) {
      * @param <R> the result value type
      * @param mapper
      *            a function that, when applied to the item emitted by the source Single, returns a SingleSource
-     * @return the Single returned from {@code func} when applied to the item emitted by the source Single
+     * @return the Single returned from {@code mapper} when applied to the item emitted by the source Single
      * @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
      */
     @SchedulerSupport(SchedulerSupport.NONE)
@@ -1826,6 +1826,28 @@ public final <R> Single<R> flatMap(Function<? super T, ? extends SingleSource<?
         return RxJavaPlugins.onAssembly(new SingleFlatMap<T, R>(this, mapper));
     }
 
+    /**
+     * Returns a Maybe that is based on applying a specified function to the item emitted by the source Single,
+     * where that function returns a MaybeSource.
+     * <p>
+     * <img width="640" height="300" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.flatMapMaybe.png" alt="">
+     * <dl>
+     * <dt><b>Scheduler:</b></dt>
+     * <dd>{@code flatMapMaybe} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param <R> the result value type
+     * @param mapper
+     *            a function that, when applied to the item emitted by the source Single, returns a MaybeSource
+     * @return the Maybe returned from {@code mapper} when applied to the item emitted by the source Single
+     * @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
+     */
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final <R> Maybe<R> flatMapMaybe(final Function<? super T, ? extends MaybeSource<? extends R>> mapper) {
+        ObjectHelper.requireNonNull(mapper, "mapper is null");
+        return RxJavaPlugins.onAssembly(new SingleFlatMapMaybe<T, R>(this, mapper));
+    }
+
     /**
      * Returns a Flowable that emits items based on applying a specified function to the item emitted by the
      * source Single, where that function returns a Publisher.
@@ -1853,7 +1875,7 @@ public final <R> Flowable<R> flatMapPublisher(Function<? super T, ? extends Publ
     }
 
     /**
-     * Returns a Single that is based on applying a specified function to the item emitted by the source Single,
+     * Returns an Observable that is based on applying a specified function to the item emitted by the source Single,
      * where that function returns a SingleSource.
      * <p>
      * <img width="640" height="300" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.flatMap.png" alt="">
@@ -1864,8 +1886,8 @@ public final <R> Flowable<R> flatMapPublisher(Function<? super T, ? extends Publ
      *
      * @param <R> the result value type
      * @param mapper
-     *            a function that, when applied to the item emitted by the source Single, returns a SingleSource
-     * @return the Single returned from {@code func} when applied to the item emitted by the source Single
+     *            a function that, when applied to the item emitted by the source Single, returns an ObservableSource
+     * @return the Observable returned from {@code func} when applied to the item emitted by the source Single
      * @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
      */
     @SchedulerSupport(SchedulerSupport.NONE)

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

@@ -0,0 +1,128 @@
+/**
+ * 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.single;
+
+import io.reactivex.Maybe;
+import io.reactivex.MaybeObserver;
+import io.reactivex.MaybeSource;
+import io.reactivex.SingleObserver;
+import io.reactivex.SingleSource;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.exceptions.Exceptions;
+import io.reactivex.functions.Function;
+import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.functions.ObjectHelper;
+import java.util.concurrent.atomic.AtomicReference;
+
+public final class SingleFlatMapMaybe<T, R> extends Maybe<R> {
+
+    final SingleSource<? extends T> source;
+
+    final Function<? super T, ? extends MaybeSource<? extends R>> mapper;
+
+    public SingleFlatMapMaybe(SingleSource<? extends T> source, Function<? super T, ? extends MaybeSource<? extends R>> mapper) {
+        this.mapper = mapper;
+        this.source = source;
+    }
+
+    @Override
+    protected void subscribeActual(MaybeObserver<? super R> actual) {
+        source.subscribe(new FlatMapSingleObserver<T, R>(actual, mapper));
+    }
+
+    static final class FlatMapSingleObserver<T, R>
+    extends AtomicReference<Disposable>
+    implements SingleObserver<T>, Disposable {
+
+        private static final long serialVersionUID = -5843758257109742742L;
+
+        final MaybeObserver<? super R> actual;
+
+        final Function<? super T, ? extends MaybeSource<? extends R>> mapper;
+
+        FlatMapSingleObserver(MaybeObserver<? super R> actual, Function<? super T, ? extends MaybeSource<? extends R>> mapper) {
+            this.actual = actual;
+            this.mapper = mapper;
+        }
+
+        @Override
+        public void dispose() {
+            DisposableHelper.dispose(this);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return DisposableHelper.isDisposed(get());
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            if (DisposableHelper.setOnce(this, d)) {
+                actual.onSubscribe(this);
+            }
+        }
+
+        @Override
+        public void onSuccess(T value) {
+            MaybeSource<? extends R> ms;
+
+            try {
+                ms = ObjectHelper.requireNonNull(mapper.apply(value), "The mapper returned a null MaybeSource");
+            } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
+                onError(ex);
+                return;
+            }
+
+            ms.subscribe(new FlatMapMaybeObserver<R>(this, actual));
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            actual.onError(e);
+        }
+    }
+
+    static final class FlatMapMaybeObserver<R> implements MaybeObserver<R> {
+
+        final AtomicReference<Disposable> parent;
+
+        final MaybeObserver<? super R> actual;
+
+        FlatMapMaybeObserver(AtomicReference<Disposable> parent, MaybeObserver<? super R> actual) {
+            this.parent = parent;
+            this.actual = actual;
+        }
+
+        @Override
+        public void onSubscribe(final Disposable d) {
+            DisposableHelper.replace(parent, d);
+        }
+
+        @Override
+        public void onSuccess(final R value) {
+            actual.onSuccess(value);
+        }
+
+        @Override
+        public void onError(final Throwable e) {
+            actual.onError(e);
+        }
+
+        @Override
+        public void onComplete() {
+            actual.onComplete();
+        }
+    }
+}

src/test/java/io/reactivex/internal/operators/single/SingleFlatMapMaybeTest.java

@@ -0,0 +1,97 @@
+/**
+ * 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.single;
+
+import io.reactivex.Maybe;
+import io.reactivex.MaybeSource;
+import io.reactivex.Single;
+import io.reactivex.functions.Function;
+import org.junit.Test;
+
+public class SingleFlatMapMaybeTest {
+    @Test(expected = NullPointerException.class)
+    public void flatMapMaybeNull() {
+        Single.just(1)
+            .flatMapMaybe(null);
+    }
+
+    @Test
+    public void flatMapMaybeValue() {
+        Single.just(1).flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
+            @Override public MaybeSource<Integer> apply(final Integer integer) throws Exception {
+                if (integer == 1) {
+                    return Maybe.just(2);
+                }
+
+                return Maybe.just(1);
+            }
+        })
+            .test()
+            .assertResult(2);
+    }
+
+    @Test
+    public void flatMapMaybeValueDifferentType() {
+        Single.just(1).flatMapMaybe(new Function<Integer, MaybeSource<String>>() {
+            @Override public MaybeSource<String> apply(final Integer integer) throws Exception {
+                if (integer == 1) {
+                    return Maybe.just("2");
+                }
+
+                return Maybe.just("1");
+            }
+        })
+            .test()
+            .assertResult("2");
+    }
+
+    @Test
+    public void flatMapMaybeValueNull() {
+        Single.just(1).flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
+            @Override public MaybeSource<Integer> apply(final Integer integer) throws Exception {
+                return null;
+            }
+        })
+            .test()
+            .assertNoValues()
+            .assertError(NullPointerException.class)
+            .assertErrorMessage("The mapper returned a null MaybeSource");
+    }
+
+    @Test
+    public void flatMapMaybeValueErrorThrown() {
+        Single.just(1).flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
+            @Override public MaybeSource<Integer> apply(final Integer integer) throws Exception {
+                throw new RuntimeException("something went terribly wrong!");
+            }
+        })
+            .test()
+            .assertNoValues()
+            .assertError(RuntimeException.class)
+            .assertErrorMessage("something went terribly wrong!");
+    }
+
+    @Test
+    public void flatMapMaybeError() {
+        RuntimeException exception = new RuntimeException("test");
+
+        Single.error(exception).flatMapMaybe(new Function<Object, MaybeSource<Object>>() {
+            @Override public MaybeSource<Object> apply(final Object integer) throws Exception {
+                return Maybe.just(new Object());
+            }
+        })
+            .test()
+            .assertError(exception);
+    }
+}