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

Issue #3892 is a goldmine for missing backpressure problems. This PR fixes the case when singleOrDefault encounters an empty source and has to emit some default value. Fixed via setting the SingleProducer on the child on termination.

Author: akarnokd

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();
+    }
 }