1.x: fix singleOrDefault() backpressure if source is empty

src/main/java/rx/internal/operators/OperatorSingle.java

@@ -16,11 +16,11 @@
 package rx.internal.operators;
 
 import java.util.NoSuchElementException;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 import rx.Observable.Operator;
-import rx.Producer;
 import rx.Subscriber;
+import rx.internal.producers.SingleProducer;
+import rx.internal.util.RxJavaPluginUtils;
 
 /**
  * If the Observable completes after emitting a single item that matches a
@@ -65,19 +65,6 @@ public Subscriber<? super T> call(final Subscriber<? super T> child) {
 
         final ParentSubscriber<T> parent = new ParentSubscriber<T>(child, hasDefaultValue,
                 defaultValue);
-
-        child.setProducer(new Producer() {
-
-            private final AtomicBoolean requestedTwo = new AtomicBoolean(false);
-
-            @Override
-            public void request(long n) {
-                if (n > 0 && requestedTwo.compareAndSet(false, true)) {
-                    parent.requestMore(2);
-                }
-            }
-
-        });
         child.add(parent);
         return parent;
     }
@@ -88,23 +75,23 @@ private static final class ParentSubscriber<T> extends Subscriber<T> {
         private final T defaultValue;
 
         private T value;
-        private boolean isNonEmpty = false;
-        private boolean hasTooManyElements = false;
+        private boolean isNonEmpty;
+        private boolean hasTooManyElements;
 
 
         ParentSubscriber(Subscriber<? super T> child, boolean hasDefaultValue,
                 T defaultValue) {
             this.child = child;
             this.hasDefaultValue = hasDefaultValue;
             this.defaultValue = defaultValue;
-        }
-
-        void requestMore(long n) {
-            request(n);
+            request(2); // could go unbounded, but test expect this
         }
 
         @Override
         public void onNext(T value) {
+            if (hasTooManyElements) {
+                return;
+            } else
             if (isNonEmpty) {
                 hasTooManyElements = true;
                 child.onError(new IllegalArgumentException("Sequence contains too many elements"));
@@ -121,12 +108,10 @@ public void onCompleted() {
                 // We have already sent an onError message
             } else {
                 if (isNonEmpty) {
-                    child.onNext(value);
-                    child.onCompleted();
+                    child.setProducer(new SingleProducer<T>(child, value));
                 } else {
                     if (hasDefaultValue) {
-                        child.onNext(defaultValue);
-                        child.onCompleted();
+                        child.setProducer(new SingleProducer<T>(child, defaultValue));
                     } else {
                         child.onError(new NoSuchElementException("Sequence contains no elements"));
                     }
@@ -136,6 +121,11 @@ public void onCompleted() {
 
         @Override
         public void onError(Throwable e) {
+            if (hasTooManyElements) {
+                RxJavaPluginUtils.handleException(e);
+                return;
+            }
+
             child.onError(e);
         }
 

src/test/java/rx/internal/operators/OperatorSingleTest.java

@@ -37,6 +37,7 @@
 import rx.functions.Action1;
 import rx.functions.Func1;
 import rx.functions.Func2;
+import rx.observers.TestSubscriber;
 
 public class OperatorSingleTest {
 
@@ -456,4 +457,19 @@ public Integer call(Integer i1, Integer i2) {
         Integer r = reduced.toBlocking().first();
         assertEquals(21, r.intValue());
     }
+
+    @Test
+    public void defaultBackpressure() {
+        TestSubscriber<Integer> ts = TestSubscriber.create(0);
+
+        Observable.<Integer>empty().singleOrDefault(1).subscribe(ts);
+
+        ts.assertNoValues();
+
+        ts.requestMore(1);
+
+        ts.assertValue(1);
+        ts.assertCompleted();
+        ts.assertNoErrors();
+    }
 }