diff --git a/src/main/java/io/reactivex/rxjava3/core/Maybe.java b/src/main/java/io/reactivex/rxjava3/core/Maybe.java
index f8cbdcdbe7..20e52f2f5f 100644
--- a/src/main/java/io/reactivex/rxjava3/core/Maybe.java
+++ b/src/main/java/io/reactivex/rxjava3/core/Maybe.java
@@ -3653,7 +3653,7 @@ public final <R> Maybe<R> flatMap(@NonNull Function<? super T, ? extends MaybeSo
     }
 
     /**
-     * Maps the {@code onSuccess}, {@code onError} or {@code onComplete} signals of this {@code Maybe} into {@link MaybeSource} and emits that
+     * Maps the {@code onSuccess}, {@code onError} or {@code onComplete} signals of the current {@code Maybe} into a {@link MaybeSource} and emits that
      * {@code MaybeSource}'s signals.
      * <p>
      * <img width="640" height="354" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.flatMap.mmm.png" alt="">
@@ -3691,7 +3691,7 @@ public final <R> Maybe<R> flatMap(
      * Returns a {@code Maybe} that emits the results of a specified function to the pair of values emitted by the
      * current {@code Maybe} and a specified mapped {@link MaybeSource}.
      * <p>
-     * <img width="640" height="390" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/mergeMap.r.png" alt="">
+     * <img width="640" height="268" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.flatMap.combiner.png" alt="">
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code flatMap} does not operate by default on a particular {@link Scheduler}.</dd>
diff --git a/src/main/java/io/reactivex/rxjava3/core/Single.java b/src/main/java/io/reactivex/rxjava3/core/Single.java
index cab4d3c073..667125b08b 100644
--- a/src/main/java/io/reactivex/rxjava3/core/Single.java
+++ b/src/main/java/io/reactivex/rxjava3/core/Single.java
@@ -3196,6 +3196,72 @@ public final <R> Single<R> flatMap(@NonNull Function<? super T, ? extends Single
         return RxJavaPlugins.onAssembly(new SingleFlatMap<>(this, mapper));
     }
 
+    /**
+     * Returns a {@code Single} that emits the results of a specified function to the pair of values emitted by the
+     * current {@code Single} and a specified mapped {@link SingleSource}.
+     * <p>
+     * <img width="640" height="268" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.flatMap.combiner.png" alt="">
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code flatMap} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param <U>
+     *            the type of items emitted by the {@code SingleSource} returned by the {@code mapper} function
+     * @param <R>
+     *            the type of items emitted by the resulting {@code Single}
+     * @param mapper
+     *            a function that returns a {@code SingleSource} for the item emitted by the current {@code Single}
+     * @param combiner
+     *            a function that combines one item emitted by each of the source and collection {@code SingleSource} and
+     *            returns an item to be emitted by the resulting {@code SingleSource}
+     * @return the new {@code Single} instance
+     * @throws NullPointerException if {@code mapper} or {@code combiner} is {@code null}
+     * @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
+     * @since 3.0.0
+     */
+    @CheckReturnValue
+    @NonNull
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final <U, R> Single<R> flatMap(@NonNull Function<? super T, ? extends SingleSource<? extends U>> mapper,
+            @NonNull BiFunction<? super T, ? super U, ? extends R> combiner) {
+        Objects.requireNonNull(mapper, "mapper is null");
+        Objects.requireNonNull(combiner, "combiner is null");
+        return RxJavaPlugins.onAssembly(new SingleFlatMapBiSelector<>(this, mapper, combiner));
+    }
+
+    /**
+     * Maps the {@code onSuccess} or {@code onError} signals of the current {@code Single} into a {@link SingleSource} and emits that
+     * {@code SingleSource}'s signals.
+     * <p>
+     * <img width="640" height="449" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.flatMap.notification.png" alt="">
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code flatMap} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param <R>
+     *            the result type
+     * @param onSuccessMapper
+     *            a function that returns a {@code SingleSource} to merge for the {@code onSuccess} item emitted by this {@code Single}
+     * @param onErrorMapper
+     *            a function that returns a {@code SingleSource} to merge for an {@code onError} notification from this {@code Single}
+     * @return the new {@code Single} instance
+     * @throws NullPointerException if {@code onSuccessMapper} or {@code onErrorMapper} is {@code null}
+     * @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
+     * @since 3.0.0
+     */
+    @CheckReturnValue
+    @NonNull
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final <R> Single<R> flatMap(
+            @NonNull Function<? super T, ? extends SingleSource<? extends R>> onSuccessMapper,
+            @NonNull Function<? super Throwable, ? extends SingleSource<? extends R>> onErrorMapper) {
+        Objects.requireNonNull(onSuccessMapper, "onSuccessMapper is null");
+        Objects.requireNonNull(onErrorMapper, "onErrorMapper is null");
+        return RxJavaPlugins.onAssembly(new SingleFlatMapNotification<>(this, onSuccessMapper, onErrorMapper));
+    }
+
     /**
      * Returns a {@link Maybe} that is based on applying a specified function to the item emitted by the current {@code Single},
      * where that function returns a {@link MaybeSource}.
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeFlatMapNotification.java b/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeFlatMapNotification.java
index e357938db2..ffe6ee5e20 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeFlatMapNotification.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeFlatMapNotification.java
@@ -109,7 +109,9 @@ public void onSuccess(T value) {
                 return;
             }
 
-            source.subscribe(new InnerObserver());
+            if (!isDisposed()) {
+                source.subscribe(new InnerObserver());
+            }
         }
 
         @Override
@@ -124,7 +126,9 @@ public void onError(Throwable e) {
                 return;
             }
 
-            source.subscribe(new InnerObserver());
+            if (!isDisposed()) {
+                source.subscribe(new InnerObserver());
+            }
         }
 
         @Override
@@ -139,7 +143,9 @@ public void onComplete() {
                 return;
             }
 
-            source.subscribe(new InnerObserver());
+            if (!isDisposed()) {
+                source.subscribe(new InnerObserver());
+            }
         }
 
         final class InnerObserver implements MaybeObserver<R> {
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeFlatMapSingle.java b/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeFlatMapSingle.java
index aed71dc855..999b22f836 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeFlatMapSingle.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeFlatMapSingle.java
@@ -89,7 +89,9 @@ public void onSuccess(T value) {
                 return;
             }
 
-            ss.subscribe(new FlatMapSingleObserver<R>(this, downstream));
+            if (!isDisposed()) {
+                ss.subscribe(new FlatMapSingleObserver<R>(this, downstream));
+            }
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/MaybeFlatMapObservable.java b/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/MaybeFlatMapObservable.java
index ae42405f74..ef1dc5f023 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/MaybeFlatMapObservable.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/MaybeFlatMapObservable.java
@@ -106,7 +106,9 @@ public void onSuccess(T t) {
                 return;
             }
 
-            o.subscribe(this);
+            if (!isDisposed()) {
+                o.subscribe(this);
+            }
         }
 
     }
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/MaybeFlatMapPublisher.java b/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/MaybeFlatMapPublisher.java
index c590755704..82a6d5da7d 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/MaybeFlatMapPublisher.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/MaybeFlatMapPublisher.java
@@ -116,7 +116,9 @@ public void onSuccess(T t) {
                 return;
             }
 
-            p.subscribe(this);
+            if (get() != SubscriptionHelper.CANCELLED) {
+                p.subscribe(this);
+            }
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/SingleFlatMapObservable.java b/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/SingleFlatMapObservable.java
index c3dd9b59c1..b2562d8dba 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/SingleFlatMapObservable.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/SingleFlatMapObservable.java
@@ -106,7 +106,9 @@ public void onSuccess(T t) {
                 return;
             }
 
-            o.subscribe(this);
+            if (!isDisposed()) {
+                o.subscribe(this);
+            }
         }
 
     }
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapBiSelector.java b/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapBiSelector.java
new file mode 100644
index 0000000000..d1051f3357
--- /dev/null
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapBiSelector.java
@@ -0,0 +1,156 @@
+/**
+ * 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.rxjava3.internal.operators.single;
+
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.reactivex.rxjava3.core.*;
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.exceptions.Exceptions;
+import io.reactivex.rxjava3.functions.*;
+import io.reactivex.rxjava3.internal.disposables.DisposableHelper;
+
+/**
+ * Maps a source item to another SingleSource then calls a BiFunction with the
+ * original item and the secondary item to generate the final result.
+ *
+ * @param <T> the main value type
+ * @param <U> the second value type
+ * @param <R> the result value type
+ * @since 3.0.0
+ */
+public final class SingleFlatMapBiSelector<T, U, R> extends Single<R> {
+
+    final SingleSource<T> source;
+
+    final Function<? super T, ? extends SingleSource<? extends U>> mapper;
+
+    final BiFunction<? super T, ? super U, ? extends R> resultSelector;
+
+    public SingleFlatMapBiSelector(SingleSource<T> source,
+            Function<? super T, ? extends SingleSource<? extends U>> mapper,
+            BiFunction<? super T, ? super U, ? extends R> resultSelector) {
+        this.source = source;
+        this.mapper = mapper;
+        this.resultSelector = resultSelector;
+    }
+
+    @Override
+    protected void subscribeActual(SingleObserver<? super R> observer) {
+        source.subscribe(new FlatMapBiMainObserver<T, U, R>(observer, mapper, resultSelector));
+    }
+
+    static final class FlatMapBiMainObserver<T, U, R>
+    implements SingleObserver<T>, Disposable {
+
+        final Function<? super T, ? extends SingleSource<? extends U>> mapper;
+
+        final InnerObserver<T, U, R> inner;
+
+        FlatMapBiMainObserver(SingleObserver<? super R> actual,
+                Function<? super T, ? extends SingleSource<? extends U>> mapper,
+                BiFunction<? super T, ? super U, ? extends R> resultSelector) {
+            this.inner = new InnerObserver<>(actual, resultSelector);
+            this.mapper = mapper;
+        }
+
+        @Override
+        public void dispose() {
+            DisposableHelper.dispose(inner);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return DisposableHelper.isDisposed(inner.get());
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            if (DisposableHelper.setOnce(inner, d)) {
+                inner.downstream.onSubscribe(this);
+            }
+        }
+
+        @Override
+        public void onSuccess(T value) {
+            SingleSource<? extends U> next;
+
+            try {
+                next = Objects.requireNonNull(mapper.apply(value), "The mapper returned a null MaybeSource");
+            } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
+                inner.downstream.onError(ex);
+                return;
+            }
+
+            if (DisposableHelper.replace(inner, null)) {
+                inner.value = value;
+                next.subscribe(inner);
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            inner.downstream.onError(e);
+        }
+
+        static final class InnerObserver<T, U, R>
+        extends AtomicReference<Disposable>
+        implements SingleObserver<U> {
+
+            private static final long serialVersionUID = -2897979525538174559L;
+
+            final SingleObserver<? super R> downstream;
+
+            final BiFunction<? super T, ? super U, ? extends R> resultSelector;
+
+            T value;
+
+            InnerObserver(SingleObserver<? super R> actual,
+                    BiFunction<? super T, ? super U, ? extends R> resultSelector) {
+                this.downstream = actual;
+                this.resultSelector = resultSelector;
+            }
+
+            @Override
+            public void onSubscribe(Disposable d) {
+                DisposableHelper.setOnce(this, d);
+            }
+
+            @Override
+            public void onSuccess(U value) {
+                T t = this.value;
+                this.value = null;
+
+                R r;
+
+                try {
+                    r = Objects.requireNonNull(resultSelector.apply(t, value), "The resultSelector returned a null value");
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    downstream.onError(ex);
+                    return;
+                }
+
+                downstream.onSuccess(r);
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                downstream.onError(e);
+            }
+        }
+    }
+}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapNotification.java b/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapNotification.java
new file mode 100644
index 0000000000..1ba9796b7c
--- /dev/null
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapNotification.java
@@ -0,0 +1,147 @@
+/**
+ * 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.rxjava3.internal.operators.single;
+
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.reactivex.rxjava3.core.*;
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.exceptions.*;
+import io.reactivex.rxjava3.functions.*;
+import io.reactivex.rxjava3.internal.disposables.DisposableHelper;
+
+/**
+ * Maps a value into a SingleSource and relays its signal.
+ *
+ * @param <T> the source value type
+ * @param <R> the result value type
+ * @since 3.0.0
+ */
+public final class SingleFlatMapNotification<T, R> extends Single<R> {
+
+    final SingleSource<T> source;
+
+    final Function<? super T, ? extends SingleSource<? extends R>> onSuccessMapper;
+
+    final Function<? super Throwable, ? extends SingleSource<? extends R>> onErrorMapper;
+
+    public SingleFlatMapNotification(SingleSource<T> source,
+            Function<? super T, ? extends SingleSource<? extends R>> onSuccessMapper,
+            Function<? super Throwable, ? extends SingleSource<? extends R>> onErrorMapper) {
+        this.source = source;
+        this.onSuccessMapper = onSuccessMapper;
+        this.onErrorMapper = onErrorMapper;
+    }
+
+    @Override
+    protected void subscribeActual(SingleObserver<? super R> observer) {
+        source.subscribe(new FlatMapSingleObserver<>(observer, onSuccessMapper, onErrorMapper));
+    }
+
+    static final class FlatMapSingleObserver<T, R>
+    extends AtomicReference<Disposable>
+    implements SingleObserver<T>, Disposable {
+
+        private static final long serialVersionUID = 4375739915521278546L;
+
+        final SingleObserver<? super R> downstream;
+
+        final Function<? super T, ? extends SingleSource<? extends R>> onSuccessMapper;
+
+        final Function<? super Throwable, ? extends SingleSource<? extends R>> onErrorMapper;
+
+        Disposable upstream;
+
+        FlatMapSingleObserver(SingleObserver<? super R> actual,
+                Function<? super T, ? extends SingleSource<? extends R>> onSuccessMapper,
+                Function<? super Throwable, ? extends SingleSource<? extends R>> onErrorMapper) {
+            this.downstream = actual;
+            this.onSuccessMapper = onSuccessMapper;
+            this.onErrorMapper = onErrorMapper;
+        }
+
+        @Override
+        public void dispose() {
+            DisposableHelper.dispose(this);
+            upstream.dispose();
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return DisposableHelper.isDisposed(get());
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            if (DisposableHelper.validate(this.upstream, d)) {
+                this.upstream = d;
+
+                downstream.onSubscribe(this);
+            }
+        }
+
+        @Override
+        public void onSuccess(T value) {
+            SingleSource<? extends R> source;
+
+            try {
+                source = Objects.requireNonNull(onSuccessMapper.apply(value), "The onSuccessMapper returned a null SingleSource");
+            } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
+                downstream.onError(ex);
+                return;
+            }
+
+            if (!isDisposed()) {
+                source.subscribe(new InnerObserver());
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            SingleSource<? extends R> source;
+
+            try {
+                source = Objects.requireNonNull(onErrorMapper.apply(e), "The onErrorMapper returned a null SingleSource");
+            } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
+                downstream.onError(new CompositeException(e, ex));
+                return;
+            }
+
+            if (!isDisposed()) {
+                source.subscribe(new InnerObserver());
+            }
+        }
+
+        final class InnerObserver implements SingleObserver<R> {
+
+            @Override
+            public void onSubscribe(Disposable d) {
+                DisposableHelper.setOnce(FlatMapSingleObserver.this, d);
+            }
+
+            @Override
+            public void onSuccess(R value) {
+                downstream.onSuccess(value);
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                downstream.onError(e);
+            }
+        }
+    }
+}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapPublisher.java b/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapPublisher.java
index f474830e40..c89913791d 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapPublisher.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapPublisher.java
@@ -92,7 +92,9 @@ public void onSuccess(S value) {
                 downstream.onError(e);
                 return;
             }
-            f.subscribe(this);
+            if (parent.get() != SubscriptionHelper.CANCELLED) {
+                f.subscribe(this);
+            }
         }
 
         @Override
diff --git a/src/test/java/io/reactivex/rxjava3/core/XFlatMapTest.java b/src/test/java/io/reactivex/rxjava3/core/XFlatMapTest.java
index 6e80b57904..06394de7ab 100644
--- a/src/test/java/io/reactivex/rxjava3/core/XFlatMapTest.java
+++ b/src/test/java/io/reactivex/rxjava3/core/XFlatMapTest.java
@@ -22,7 +22,7 @@
 import org.reactivestreams.Publisher;
 
 import io.reactivex.rxjava3.exceptions.TestException;
-import io.reactivex.rxjava3.functions.Function;
+import io.reactivex.rxjava3.functions.*;
 import io.reactivex.rxjava3.observers.TestObserver;
 import io.reactivex.rxjava3.plugins.RxJavaPlugins;
 import io.reactivex.rxjava3.schedulers.Schedulers;
@@ -224,7 +224,7 @@ public Completable apply(Integer v) throws Exception {
     }
 
     @Test
-    public void observableFlowable() throws Exception {
+    public void observableObservable() throws Exception {
         List<Throwable> errors = TestHelper.trackPluginErrors();
         try {
             TestObserver<Integer> to = Observable.just(1)
@@ -505,7 +505,179 @@ public Completable apply(Integer v) throws Exception {
     }
 
     @Test
-    @Ignore
+    public void singlePublisher() throws Exception {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            TestSubscriber<Integer> ts = Single.just(1)
+            .subscribeOn(Schedulers.io())
+            .flatMapPublisher(new Function<Integer, Publisher<Integer>>() {
+                @Override
+                public Publisher<Integer> apply(Integer v) throws Exception {
+                    sleep();
+                    return Flowable.<Integer>error(new TestException());
+                }
+            })
+            .test();
+
+            cb.await();
+
+            beforeCancelSleep(ts);
+
+            ts.cancel();
+
+            Thread.sleep(SLEEP_AFTER_CANCEL);
+
+            ts.assertEmpty();
+
+            assertTrue(errors.toString(), errors.isEmpty());
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
+    @Test
+    public void singleCombiner() throws Exception {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            TestObserver<Integer> to = Single.just(1)
+            .subscribeOn(Schedulers.io())
+            .flatMap(new Function<Integer, Single<Integer>>() {
+                @Override
+                public Single<Integer> apply(Integer v) throws Exception {
+                    sleep();
+                    return Single.<Integer>error(new TestException());
+                }
+            }, (a, b) -> a + b)
+            .test();
+
+            cb.await();
+
+            beforeCancelSleep(to);
+
+            to.dispose();
+
+            Thread.sleep(SLEEP_AFTER_CANCEL);
+
+            to.assertEmpty();
+
+            assertTrue(errors.toString(), errors.isEmpty());
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
+    @Test
+    public void singleObservable() throws Exception {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            TestObserver<Integer> to = Single.just(1)
+            .subscribeOn(Schedulers.io())
+            .flatMapObservable(new Function<Integer, Observable<Integer>>() {
+                @Override
+                public Observable<Integer> apply(Integer v) throws Exception {
+                    sleep();
+                    return Observable.<Integer>error(new TestException());
+                }
+            })
+            .test();
+
+            cb.await();
+
+            beforeCancelSleep(to);
+
+            to.dispose();
+
+            Thread.sleep(SLEEP_AFTER_CANCEL);
+
+            to.assertEmpty();
+
+            assertTrue(errors.toString(), errors.isEmpty());
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
+    @Test
+    public void singleNotificationSuccess() throws Exception {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            TestObserver<Integer> to = Single.just(1)
+            .subscribeOn(Schedulers.io())
+            .flatMap(
+                new Function<Integer, Single<Integer>>() {
+                    @Override
+                    public Single<Integer> apply(Integer v) throws Exception {
+                        sleep();
+                        return Single.<Integer>error(new TestException());
+                    }
+                },
+                new Function<Throwable, Single<Integer>>() {
+                    @Override
+                    public Single<Integer> apply(Throwable v) throws Exception {
+                        sleep();
+                        return Single.<Integer>error(new TestException());
+                    }
+                }
+            )
+            .test();
+
+            cb.await();
+
+            beforeCancelSleep(to);
+
+            to.dispose();
+
+            Thread.sleep(SLEEP_AFTER_CANCEL);
+
+            to.assertEmpty();
+
+            assertTrue(errors.toString(), errors.isEmpty());
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
+    @Test
+    public void singleNotificationError() throws Exception {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            TestObserver<Integer> to = Single.<Integer>error(new TestException())
+            .subscribeOn(Schedulers.io())
+            .flatMap(
+                new Function<Integer, Single<Integer>>() {
+                    @Override
+                    public Single<Integer> apply(Integer v) throws Exception {
+                        sleep();
+                        return Single.<Integer>error(new TestException());
+                    }
+                },
+                new Function<Throwable, Single<Integer>>() {
+                    @Override
+                    public Single<Integer> apply(Throwable v) throws Exception {
+                        sleep();
+                        return Single.<Integer>error(new TestException());
+                    }
+                }
+            )
+            .test();
+
+            cb.await();
+
+            beforeCancelSleep(to);
+
+            to.dispose();
+
+            Thread.sleep(SLEEP_AFTER_CANCEL);
+
+            to.assertEmpty();
+
+            assertTrue(errors.toString(), errors.isEmpty());
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
+    @Test
     public void maybeSingle() throws Exception {
         List<Throwable> errors = TestHelper.trackPluginErrors();
         try {
@@ -537,6 +709,37 @@ public Single<Integer> apply(Integer v) throws Exception {
         }
     }
 
+    @Test
+    public void maybeSingle2() throws Exception {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            TestObserver<Integer> to = Maybe.just(1)
+            .subscribeOn(Schedulers.io())
+            .flatMapSingle(new Function<Integer, Single<Integer>>() {
+                @Override
+                public Single<Integer> apply(Integer v) throws Exception {
+                    sleep();
+                    return Single.<Integer>error(new TestException());
+                }
+            })
+            .test();
+
+            cb.await();
+
+            beforeCancelSleep(to);
+
+            to.dispose();
+
+            Thread.sleep(SLEEP_AFTER_CANCEL);
+
+            to.assertEmpty();
+
+            assertTrue(errors.toString(), errors.isEmpty());
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
     @Test
     public void maybeMaybe() throws Exception {
         List<Throwable> errors = TestHelper.trackPluginErrors();
@@ -568,6 +771,240 @@ public Maybe<Integer> apply(Integer v) throws Exception {
         }
     }
 
+    @Test
+    public void maybePublisher() throws Exception {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            TestSubscriber<Integer> ts = Maybe.just(1)
+            .subscribeOn(Schedulers.io())
+            .flatMapPublisher(new Function<Integer, Publisher<Integer>>() {
+                @Override
+                public Publisher<Integer> apply(Integer v) throws Exception {
+                    sleep();
+                    return Flowable.<Integer>error(new TestException());
+                }
+            })
+            .test();
+
+            cb.await();
+
+            beforeCancelSleep(ts);
+
+            ts.cancel();
+
+            Thread.sleep(SLEEP_AFTER_CANCEL);
+
+            ts.assertEmpty();
+
+            assertTrue(errors.toString(), errors.isEmpty());
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
+    @Test
+    public void maybeObservable() throws Exception {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            TestObserver<Integer> to = Maybe.just(1)
+            .subscribeOn(Schedulers.io())
+            .flatMapObservable(new Function<Integer, Observable<Integer>>() {
+                @Override
+                public Observable<Integer> apply(Integer v) throws Exception {
+                    sleep();
+                    return Observable.<Integer>error(new TestException());
+                }
+            })
+            .test();
+
+            cb.await();
+
+            beforeCancelSleep(to);
+
+            to.dispose();
+
+            Thread.sleep(SLEEP_AFTER_CANCEL);
+
+            to.assertEmpty();
+
+            assertTrue(errors.toString(), errors.isEmpty());
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
+    @Test
+    public void maybeNotificationSuccess() throws Exception {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            TestObserver<Integer> to = Maybe.just(1)
+            .subscribeOn(Schedulers.io())
+            .flatMap(
+                new Function<Integer, Maybe<Integer>>() {
+                    @Override
+                    public Maybe<Integer> apply(Integer v) throws Exception {
+                        sleep();
+                        return Maybe.<Integer>error(new TestException());
+                    }
+                },
+                new Function<Throwable, Maybe<Integer>>() {
+                    @Override
+                    public Maybe<Integer> apply(Throwable v) throws Exception {
+                        sleep();
+                        return Maybe.<Integer>error(new TestException());
+                    }
+                },
+                new Supplier<Maybe<Integer>>() {
+                    @Override
+                    public Maybe<Integer> get() throws Exception {
+                        sleep();
+                        return Maybe.<Integer>error(new TestException());
+                    }
+                }
+            )
+            .test();
+
+            cb.await();
+
+            beforeCancelSleep(to);
+
+            to.dispose();
+
+            Thread.sleep(SLEEP_AFTER_CANCEL);
+
+            to.assertEmpty();
+
+            assertTrue(errors.toString(), errors.isEmpty());
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
+    @Test
+    public void maybeNotificationError() throws Exception {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            TestObserver<Integer> to = Maybe.<Integer>error(new TestException())
+            .subscribeOn(Schedulers.io())
+            .flatMap(
+                new Function<Integer, Maybe<Integer>>() {
+                    @Override
+                    public Maybe<Integer> apply(Integer v) throws Exception {
+                        sleep();
+                        return Maybe.<Integer>error(new TestException());
+                    }
+                },
+                new Function<Throwable, Maybe<Integer>>() {
+                    @Override
+                    public Maybe<Integer> apply(Throwable v) throws Exception {
+                        sleep();
+                        return Maybe.<Integer>error(new TestException());
+                    }
+                },
+                new Supplier<Maybe<Integer>>() {
+                    @Override
+                    public Maybe<Integer> get() throws Exception {
+                        sleep();
+                        return Maybe.<Integer>error(new TestException());
+                    }
+                }
+            )
+            .test();
+
+            cb.await();
+
+            beforeCancelSleep(to);
+
+            to.dispose();
+
+            Thread.sleep(SLEEP_AFTER_CANCEL);
+
+            to.assertEmpty();
+
+            assertTrue(errors.toString(), errors.isEmpty());
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
+    @Test
+    public void maybeNotificationEmpty() throws Exception {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            TestObserver<Integer> to = Maybe.<Integer>empty()
+            .subscribeOn(Schedulers.io())
+            .flatMap(
+                new Function<Integer, Maybe<Integer>>() {
+                    @Override
+                    public Maybe<Integer> apply(Integer v) throws Exception {
+                        sleep();
+                        return Maybe.<Integer>error(new TestException());
+                    }
+                },
+                new Function<Throwable, Maybe<Integer>>() {
+                    @Override
+                    public Maybe<Integer> apply(Throwable v) throws Exception {
+                        sleep();
+                        return Maybe.<Integer>error(new TestException());
+                    }
+                },
+                new Supplier<Maybe<Integer>>() {
+                    @Override
+                    public Maybe<Integer> get() throws Exception {
+                        sleep();
+                        return Maybe.<Integer>error(new TestException());
+                    }
+                }
+            )
+            .test();
+
+            cb.await();
+
+            beforeCancelSleep(to);
+
+            to.dispose();
+
+            Thread.sleep(SLEEP_AFTER_CANCEL);
+
+            to.assertEmpty();
+
+            assertTrue(errors.toString(), errors.isEmpty());
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
+    @Test
+    public void maybeCombiner() throws Exception {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            TestObserver<Integer> to = Maybe.just(1)
+            .subscribeOn(Schedulers.io())
+            .flatMap(new Function<Integer, Maybe<Integer>>() {
+                @Override
+                public Maybe<Integer> apply(Integer v) throws Exception {
+                    sleep();
+                    return Maybe.<Integer>error(new TestException());
+                }
+            }, (a, b) -> a + b)
+            .test();
+
+            cb.await();
+
+            beforeCancelSleep(to);
+
+            to.dispose();
+
+            Thread.sleep(SLEEP_AFTER_CANCEL);
+
+            to.assertEmpty();
+
+            assertTrue(errors.toString(), errors.isEmpty());
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
     @Test
     public void maybeCompletable() throws Exception {
         List<Throwable> errors = TestHelper.trackPluginErrors();
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapBiSelectorTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapBiSelectorTest.java
new file mode 100644
index 0000000000..84a2a650c4
--- /dev/null
+++ b/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapBiSelectorTest.java
@@ -0,0 +1,205 @@
+/**
+ * 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.rxjava3.internal.operators.single;
+
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Test;
+
+import io.reactivex.rxjava3.core.*;
+import io.reactivex.rxjava3.exceptions.TestException;
+import io.reactivex.rxjava3.functions.*;
+import io.reactivex.rxjava3.observers.TestObserver;
+import io.reactivex.rxjava3.subjects.SingleSubject;
+import io.reactivex.rxjava3.testsupport.TestHelper;
+
+public class SingleFlatMapBiSelectorTest extends RxJavaTest {
+
+    BiFunction<Integer, Integer, String> stringCombine() {
+        return new BiFunction<Integer, Integer, String>() {
+            @Override
+            public String apply(Integer a, Integer b) throws Exception {
+                return a + ":" + b;
+            }
+        };
+    }
+
+    @Test
+    public void normal() {
+        Single.just(1)
+        .flatMap(new Function<Integer, SingleSource<Integer>>() {
+            @Override
+            public SingleSource<Integer> apply(Integer v) throws Exception {
+                return Single.just(2);
+            }
+        }, stringCombine())
+        .test()
+        .assertResult("1:2");
+    }
+
+    @Test
+    public void errorWithJust() {
+        final int[] call = { 0 };
+
+        Single.<Integer>error(new TestException())
+        .flatMap(new Function<Integer, SingleSource<Integer>>() {
+            @Override
+            public SingleSource<Integer> apply(Integer v) throws Exception {
+                call[0]++;
+                return Single.just(1);
+            }
+        }, stringCombine())
+        .test()
+        .assertFailure(TestException.class);
+
+        assertEquals(0, call[0]);
+    }
+
+    @Test
+    public void justWithError() {
+        final int[] call = { 0 };
+
+        Single.just(1)
+        .flatMap(new Function<Integer, SingleSource<Integer>>() {
+            @Override
+            public SingleSource<Integer> apply(Integer v) throws Exception {
+                call[0]++;
+                return Single.<Integer>error(new TestException());
+            }
+        }, stringCombine())
+        .test()
+        .assertFailure(TestException.class);
+
+        assertEquals(1, call[0]);
+    }
+
+    @Test
+    public void dispose() {
+        TestHelper.checkDisposed(SingleSubject.create()
+                .flatMap(new Function<Object, SingleSource<Integer>>() {
+            @Override
+            public SingleSource<Integer> apply(Object v) throws Exception {
+                return Single.just(1);
+            }
+        }, new BiFunction<Object, Integer, Object>() {
+            @Override
+            public Object apply(Object a, Integer b) throws Exception {
+                return b;
+            }
+        }));
+    }
+
+    @Test
+    public void doubleOnSubscribe() {
+        TestHelper.checkDoubleOnSubscribeSingle(new Function<Single<Object>, SingleSource<Object>>() {
+            @Override
+            public SingleSource<Object> apply(Single<Object> v) throws Exception {
+                return v.flatMap(new Function<Object, SingleSource<Integer>>() {
+                    @Override
+                    public SingleSource<Integer> apply(Object v) throws Exception {
+                        return Single.just(1);
+                    }
+                }, new BiFunction<Object, Integer, Object>() {
+                    @Override
+                    public Object apply(Object a, Integer b) throws Exception {
+                        return b;
+                    }
+                });
+            }
+        });
+    }
+
+    @Test
+    public void mapperThrows() {
+        Single.just(1)
+        .flatMap(new Function<Integer, SingleSource<Integer>>() {
+            @Override
+            public SingleSource<Integer> apply(Integer v) throws Exception {
+                throw new TestException();
+            }
+        }, stringCombine())
+        .test()
+        .assertFailure(TestException.class);
+    }
+
+    @Test
+    public void mapperReturnsNull() {
+        Single.just(1)
+        .flatMap(new Function<Integer, SingleSource<Integer>>() {
+            @Override
+            public SingleSource<Integer> apply(Integer v) throws Exception {
+                return null;
+            }
+        }, stringCombine())
+        .test()
+        .assertFailure(NullPointerException.class);
+    }
+
+    @Test
+    public void resultSelectorThrows() {
+        Single.just(1)
+        .flatMap(new Function<Integer, SingleSource<Integer>>() {
+            @Override
+            public SingleSource<Integer> apply(Integer v) throws Exception {
+                return Single.just(2);
+            }
+        }, new BiFunction<Integer, Integer, Object>() {
+            @Override
+            public Object apply(Integer a, Integer b) throws Exception {
+                throw new TestException();
+            }
+        })
+        .test()
+        .assertFailure(TestException.class);
+    }
+
+    @Test
+    public void resultSelectorReturnsNull() {
+        Single.just(1)
+        .flatMap(new Function<Integer, SingleSource<Integer>>() {
+            @Override
+            public SingleSource<Integer> apply(Integer v) throws Exception {
+                return Single.just(2);
+            }
+        }, new BiFunction<Integer, Integer, Object>() {
+            @Override
+            public Object apply(Integer a, Integer b) throws Exception {
+                return null;
+            }
+        })
+        .test()
+        .assertFailure(NullPointerException.class);
+    }
+
+    @Test
+    public void mapperCancels() {
+        final TestObserver<Integer> to = new TestObserver<>();
+
+        Single.just(1)
+        .flatMap(new Function<Integer, SingleSource<Integer>>() {
+            @Override
+            public SingleSource<Integer> apply(Integer v) throws Exception {
+                to.dispose();
+                return Single.just(2);
+            }
+        }, new BiFunction<Integer, Integer, Integer>() {
+            @Override
+            public Integer apply(Integer a, Integer b) throws Exception {
+                throw new IllegalStateException();
+            }
+        })
+        .subscribeWith(to)
+        .assertEmpty();
+    }
+}
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapNotificationTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapNotificationTest.java
new file mode 100644
index 0000000000..266acc93f2
--- /dev/null
+++ b/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleFlatMapNotificationTest.java
@@ -0,0 +1,113 @@
+/**
+ * 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.rxjava3.internal.operators.single;
+
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.junit.Test;
+
+import io.reactivex.rxjava3.core.*;
+import io.reactivex.rxjava3.exceptions.*;
+import io.reactivex.rxjava3.functions.Function;
+import io.reactivex.rxjava3.internal.functions.Functions;
+import io.reactivex.rxjava3.testsupport.*;
+
+public class SingleFlatMapNotificationTest extends RxJavaTest {
+
+    @Test
+    public void dispose() {
+        TestHelper.checkDisposed(Single.just(1)
+                .flatMap(Functions.justFunction(Single.just(1)),
+                        Functions.justFunction(Single.just(1))));
+    }
+
+    @Test
+    public void doubleOnSubscribe() {
+        TestHelper.checkDoubleOnSubscribeSingle(new Function<Single<Integer>, SingleSource<Integer>>() {
+            @Override
+            public SingleSource<Integer> apply(Single<Integer> m) throws Exception {
+                return m
+                        .flatMap(Functions.justFunction(Single.just(1)),
+                                Functions.justFunction(Single.just(1)));
+            }
+        });
+    }
+
+    @Test
+    public void onSuccessNull() {
+        Single.just(1)
+        .flatMap(Functions.justFunction((Single<Integer>)null),
+                Functions.justFunction(Single.just(1)))
+        .test()
+        .assertFailure(NullPointerException.class);
+    }
+
+    @Test
+    public void onErrorNull() {
+        TestObserverEx<Integer> to = Single.<Integer>error(new TestException())
+        .flatMap(Functions.justFunction(Single.just(1)),
+                Functions.justFunction((Single<Integer>)null))
+        .to(TestHelper.<Integer>testConsumer())
+        .assertFailure(CompositeException.class);
+
+        List<Throwable> ce = TestHelper.compositeList(to.errors().get(0));
+
+        TestHelper.assertError(ce, 0, TestException.class);
+        TestHelper.assertError(ce, 1, NullPointerException.class);
+    }
+
+    @Test
+    public void onSuccessError() {
+        Single.just(1)
+        .flatMap(Functions.justFunction(Single.<Integer>error(new TestException())),
+                Functions.justFunction((Single<Integer>)null))
+        .test()
+        .assertFailure(TestException.class);
+    }
+
+    @Test
+    public void onSucccessSuccess() {
+        Single.just(1)
+        .flatMap(v -> Single.just(2), e -> Single.just(3))
+        .test()
+        .assertResult(2);
+    }
+
+    @Test
+    public void onErrorSuccess() throws Throwable {
+        TestHelper.withErrorTracking(errors -> {
+            Single.error(new TestException())
+            .flatMap(v -> Single.just(2), e -> Single.just(3))
+            .test()
+            .assertResult(3);
+
+            assertTrue("" + errors, errors.isEmpty());
+        });
+    }
+
+    @Test
+    public void onErrorError() throws Throwable {
+        TestHelper.withErrorTracking(errors -> {
+            Single.error(new TestException())
+            .flatMap(v -> Single.just(2), e -> Single.<Integer>error(new IOException()))
+            .test()
+            .assertFailure(IOException.class);
+
+            assertTrue("" + errors, errors.isEmpty());
+        });
+    }
+}