add backpressure support for defaultIfEmpty()

Author: davidmoten

src/main/java/rx/Observable.java

@@ -19,6 +19,7 @@
 import rx.exceptions.*;
 import rx.functions.*;
 import rx.internal.operators.*;
+import rx.internal.producers.SingleProducer;
 import rx.internal.util.*;
 import rx.observables.*;
 import rx.observers.SafeSubscriber;
@@ -3857,8 +3858,14 @@ public final Observable<T> debounce(long timeout, TimeUnit unit, Scheduler sched
      *         items, or the items emitted by the source Observable
      * @see <a href="http://reactivex.io/documentation/operators/defaultifempty.html">ReactiveX operators documentation: DefaultIfEmpty</a>
      */
-    public final Observable<T> defaultIfEmpty(T defaultValue) {
-        return lift(new OperatorDefaultIfEmpty<T>(defaultValue));
+    public final Observable<T> defaultIfEmpty(final T defaultValue) {
+        //if empty switch to an observable that emits defaultValue and supports backpressure
+        return switchIfEmpty(Observable.create(new OnSubscribe<T>() {
+
+            @Override
+            public void call(Subscriber<? super T> subscriber) {
+                subscriber.setProducer(new SingleProducer<T>(subscriber, defaultValue));
+            }}));
     }
 
     /**

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

@@ -26,6 +26,7 @@
 import rx.Observer;
 import rx.Subscriber;
 import rx.exceptions.TestException;
+import rx.observers.TestSubscriber;
 
 public class OperatorDefaultIfEmptyTest {
 
@@ -85,4 +86,28 @@ public void onCompleted() {
         verify(o, never()).onNext(any(Integer.class));
         verify(o, never()).onCompleted();
     }
+    
+    @Test
+    public void testBackpressureEmpty() {
+        TestSubscriber<Integer> ts = TestSubscriber.create(0);
+        Observable.<Integer>empty().defaultIfEmpty(1).subscribe(ts);
+        ts.assertNoValues();
+        ts.assertNoTerminalEvent();
+        ts.requestMore(1);
+        ts.assertValue(1);
+        ts.assertCompleted();
+    }
+    
+    @Test
+    public void testBackpressureNonEmpty() {
+        TestSubscriber<Integer> ts = TestSubscriber.create(0);
+        Observable.just(1,2,3).defaultIfEmpty(1).subscribe(ts);
+        ts.assertNoValues();
+        ts.assertNoTerminalEvent();
+        ts.requestMore(2);
+        ts.assertValues(1, 2);
+        ts.requestMore(1);
+        ts.assertValues(1, 2, 3);
+        ts.assertCompleted();
+    }
 }