2.x: add full implementation for Single.flatMapPublisher so doesn't batch requests (#6015)

Author: davidmoten

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

@@ -2460,7 +2460,8 @@ public final <R> Maybe<R> flatMapMaybe(final Function<? super T, ? extends Maybe
     @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <R> Flowable<R> flatMapPublisher(Function<? super T, ? extends Publisher<? extends R>> mapper) {
-        return toFlowable().flatMap(mapper);
+        ObjectHelper.requireNonNull(mapper, "mapper is null");
+        return RxJavaPlugins.onAssembly(new SingleFlatMapPublisher<T, R>(this, mapper));
     }
 
     /**

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

@@ -0,0 +1,115 @@
+/**
+ * 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.single;
+
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+import io.reactivex.Flowable;
+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.functions.ObjectHelper;
+import io.reactivex.internal.subscriptions.SubscriptionHelper;
+
+public final class SingleFlatMapPublisher<S, T> extends Flowable<T> {
+
+    final SingleSource<S> source;
+    final Function<? super S, ? extends Publisher<? extends T>> mapper;
+
+    public SingleFlatMapPublisher(SingleSource<S> source,
+            Function<? super S, ? extends Publisher<? extends T>> mapper) {
+        this.source = source;
+        this.mapper = mapper;
+    }
+
+    @Override
+    protected void subscribeActual(Subscriber<? super T> actual) {
+        source.subscribe(new SingleFlatMapPublisherObserver<S, T>(actual, mapper));
+    }
+
+    static final class SingleFlatMapPublisherObserver<S, T> extends AtomicLong
+            implements SingleObserver<S>, Subscriber<T>, Subscription {
+
+        private static final long serialVersionUID = 7759721921468635667L;
+
+        final Subscriber<? super T> actual;
+        final Function<? super S, ? extends Publisher<? extends T>> mapper;
+        final AtomicReference<Subscription> parent = new AtomicReference<Subscription>();
+        Disposable disposable;
+
+        SingleFlatMapPublisherObserver(Subscriber<? super T> actual,
+                Function<? super S, ? extends Publisher<? extends T>> mapper) {
+            this.actual = actual;
+            this.mapper = mapper;
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            this.disposable = d;
+            actual.onSubscribe(this);
+        }
+
+        @Override
+        public void onSuccess(S value) {
+            Publisher<? extends T> f;
+            try {
+                f = ObjectHelper.requireNonNull(mapper.apply(value), "mapper returns null");
+            } catch (Exception e) {
+                Exceptions.throwIfFatal(e);
+                actual.onError(e);
+                return;
+            }
+            f.subscribe(this);
+        }
+
+        @Override
+        public void onSubscribe(Subscription s) {
+            SubscriptionHelper.deferredSetOnce(parent, this, s);
+        }
+
+        @Override
+        public void onNext(T t) {
+            actual.onNext(t);
+        }
+
+        @Override
+        public void onComplete() {
+            actual.onComplete();
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            actual.onError(e);
+        }
+
+        @Override
+        public void request(long n) {
+            SubscriptionHelper.deferredRequest(parent, this, n);
+        }
+
+        @Override
+        public void cancel() {
+            disposable.dispose();
+            SubscriptionHelper.cancel(parent);
+        }
+    }
+
+}

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

@@ -15,12 +15,15 @@
 
 import static org.junit.Assert.*;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+
 import org.junit.Test;
 import org.reactivestreams.Publisher;
 
 import io.reactivex.*;
 import io.reactivex.exceptions.TestException;
 import io.reactivex.functions.*;
+import io.reactivex.subscribers.TestSubscriber;
 
 public class SingleFlatMapTest {
 
@@ -126,6 +129,92 @@ public Publisher<Integer> apply(Integer v) throws Exception {
         .test()
         .assertResult(1, 2, 3, 4, 5);
     }
+    
+    @Test(expected = NullPointerException.class)
+    public void flatMapPublisherMapperNull() {
+        Single.just(1).flatMapPublisher(null);
+    }
+    
+    @Test
+    public void flatMapPublisherMapperThrows() {
+        final TestException ex = new TestException();
+        Single.just(1) 
+        .flatMapPublisher(new Function<Integer, Publisher<Integer>>() {
+            @Override
+            public Publisher<Integer> apply(Integer v) throws Exception {
+                throw ex;
+            }
+        })
+        .test()
+        .assertNoValues()
+        .assertError(ex);
+    }
+    
+    @Test
+    public void flatMapPublisherSingleError() {
+        final TestException ex = new TestException();
+        Single.<Integer>error(ex) 
+        .flatMapPublisher(new Function<Integer, Publisher<Integer>>() {
+            @Override
+            public Publisher<Integer> apply(Integer v) throws Exception {
+                return Flowable.just(1);
+            }
+        })
+        .test()
+        .assertNoValues()
+        .assertError(ex);
+    }
+    
+    @Test
+    public void flatMapPublisherCancelDuringSingle() {
+        final AtomicBoolean disposed = new AtomicBoolean();
+        TestSubscriber<Integer> ts = Single.<Integer>never()
+        .doOnDispose(new Action() {
+            @Override
+            public void run() throws Exception {
+                disposed.set(true);
+            }
+        })
+        .flatMapPublisher(new Function<Integer, Publisher<Integer>>() {
+            @Override
+            public Publisher<Integer> apply(Integer v) throws Exception {
+                return Flowable.range(v, 5);
+            }
+        })
+        .test()
+        .assertNoValues() 
+        .assertNotTerminated();
+        assertFalse(disposed.get());
+        ts.cancel();
+        assertTrue(disposed.get());
+        ts.assertNotTerminated();
+    }
+    
+    @Test
+    public void flatMapPublisherCancelDuringFlowable() {
+        final AtomicBoolean disposed = new AtomicBoolean();
+        TestSubscriber<Integer> ts = 
+        Single.just(1)
+        .flatMapPublisher(new Function<Integer, Publisher<Integer>>() {
+            @Override
+            public Publisher<Integer> apply(Integer v) throws Exception {
+                return Flowable.<Integer>never()
+                        .doOnCancel(new Action() {
+                            @Override
+                            public void run() throws Exception {
+                                disposed.set(true);
+                            }
+                        });
+            }
+        })
+        .test()
+        .assertNoValues() 
+        .assertNotTerminated();
+        assertFalse(disposed.get());
+        ts.cancel();
+        assertTrue(disposed.get());
+        ts.assertNotTerminated();
+    }
 
     @Test(expected = NullPointerException.class)
     public void flatMapNull() {