add backpressure support for defaultIfEmpty()

Author: davidmoten

src/main/java/rx/Observable.java

@@ -3858,7 +3858,8 @@ public final Observable<T> debounce(long timeout, TimeUnit unit, Scheduler sched
      * @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));
+        //if empty switch to an observable that emits defaultValue and supports backpressure
+        return switchIfEmpty(Observable.from((Arrays.asList(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();
+    }
 }