2.x: Fix Flowable.singleOrError().toFlowable() not signalling NSEE (#5904)

Author: akarnokd

src/main/java/io/reactivex/internal/operators/flowable/FlowableSingle.java

@@ -13,6 +13,8 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import java.util.NoSuchElementException;
+
 import org.reactivestreams.*;
 
 import io.reactivex.*;
@@ -23,14 +25,17 @@ public final class FlowableSingle<T> extends AbstractFlowableWithUpstream<T, T>
 
     final T defaultValue;
 
-    public FlowableSingle(Flowable<T> source, T defaultValue) {
+    final boolean failOnEmpty;
+
+    public FlowableSingle(Flowable<T> source, T defaultValue, boolean failOnEmpty) {
         super(source);
         this.defaultValue = defaultValue;
+        this.failOnEmpty = failOnEmpty;
     }
 
     @Override
     protected void subscribeActual(Subscriber<? super T> s) {
-        source.subscribe(new SingleElementSubscriber<T>(s, defaultValue));
+        source.subscribe(new SingleElementSubscriber<T>(s, defaultValue, failOnEmpty));
     }
 
     static final class SingleElementSubscriber<T> extends DeferredScalarSubscription<T>
@@ -40,13 +45,16 @@ static final class SingleElementSubscriber<T> extends DeferredScalarSubscription
 
         final T defaultValue;
 
+        final boolean failOnEmpty;
+
         Subscription s;
 
         boolean done;
 
-        SingleElementSubscriber(Subscriber<? super T> actual, T defaultValue) {
+        SingleElementSubscriber(Subscriber<? super T> actual, T defaultValue, boolean failOnEmpty) {
             super(actual);
             this.defaultValue = defaultValue;
+            this.failOnEmpty = failOnEmpty;
         }
 
         @Override
@@ -94,7 +102,11 @@ public void onComplete() {
                 v = defaultValue;
             }
             if (v == null) {
-                actual.onComplete();
+                if (failOnEmpty) {
+                    actual.onError(new NoSuchElementException());
+                } else {
+                    actual.onComplete();
+                }
             } else {
                 complete(v);
             }

src/main/java/io/reactivex/internal/operators/flowable/FlowableSingleMaybe.java

@@ -36,7 +36,7 @@ protected void subscribeActual(MaybeObserver<? super T> s) {
 
     @Override
     public Flowable<T> fuseToFlowable() {
-        return RxJavaPlugins.onAssembly(new FlowableSingle<T>(source, null));
+        return RxJavaPlugins.onAssembly(new FlowableSingle<T>(source, null, false));
     }
 
     static final class SingleElementSubscriber<T>

src/main/java/io/reactivex/internal/operators/flowable/FlowableSingleSingle.java

@@ -41,7 +41,7 @@ protected void subscribeActual(SingleObserver<? super T> s) {
 
     @Override
     public Flowable<T> fuseToFlowable() {
-        return RxJavaPlugins.onAssembly(new FlowableSingle<T>(source, defaultValue));
+        return RxJavaPlugins.onAssembly(new FlowableSingle<T>(source, defaultValue, true));
     }
 
     static final class SingleElementSubscriber<T>

src/test/java/io/reactivex/internal/operators/flowable/FlowableSingleTest.java

@@ -796,4 +796,13 @@ public void cancelAsFlowable() {
 
         assertFalse(pp.hasSubscribers());
     }
+
+    @Test
+    public void singleOrError() {
+        Flowable.empty()
+        .singleOrError()
+        .toFlowable()
+        .test()
+        .assertFailure(NoSuchElementException.class);
+    }
 }

src/test/java/io/reactivex/internal/operators/observable/ObservableSingleTest.java

@@ -556,4 +556,13 @@ public MaybeSource<Object> apply(Observable<Object> o) throws Exception {
             }
         });
     }
+
+    @Test
+    public void singleOrError() {
+        Observable.empty()
+        .singleOrError()
+        .toObservable()
+        .test()
+        .assertFailure(NoSuchElementException.class);
+    }
 }