Merge branch 'OperatorDefaultIfEmpty' of github.com:akarnokd/RxJava into merge-prs

Conflicts:
	rxjava-core/src/main/java/rx/Observable.java

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -49,7 +49,6 @@
 import rx.operators.OnSubscribeFromIterable;
 import rx.operators.OnSubscribeRange;
 import rx.operators.OperationDebounce;
-import rx.operators.OperationDefaultIfEmpty;
 import rx.operators.OperationDefer;
 import rx.operators.OperationDelay;
 import rx.operators.OperationDematerialize;
@@ -98,6 +97,7 @@
 import rx.operators.OperatorCast;
 import rx.operators.OperatorCombineLatest;
 import rx.operators.OperatorConcat;
+import rx.operators.OperatorDefaultIfEmpty;
 import rx.operators.OperatorDoOnEach;
 import rx.operators.OperatorElementAt;
 import rx.operators.OperatorFilter;
@@ -3469,7 +3469,7 @@ public final Observable<T> debounce(long timeout, TimeUnit unit, Scheduler sched
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229624.aspx">MSDN: Observable.DefaultIfEmpty</a>
      */
     public final Observable<T> defaultIfEmpty(T defaultValue) {
-        return create(OperationDefaultIfEmpty.defaultIfEmpty(this, defaultValue));
+        return lift(new OperatorDefaultIfEmpty<T>(defaultValue));
     }
 
     /**

rxjava-core/src/main/java/rx/operators/OperationDefaultIfEmpty.java

@@ -0,0 +1,64 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package rx.operators;
+
+import rx.Observable.Operator;
+import rx.Subscriber;
+
+/**
+ * Returns the elements of the specified sequence or the specified default value
+ * in a singleton sequence if the sequence is empty.
+ * @param <T> the value type
+ */
+public class OperatorDefaultIfEmpty<T> implements Operator<T, T> {
+    final T defaultValue;
+
+    public OperatorDefaultIfEmpty(T defaultValue) {
+        this.defaultValue = defaultValue;
+    }
+    
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super T> child) {
+        return new Subscriber<T>(child) {
+            boolean hasValue;
+            @Override
+            public void onNext(T t) {
+                hasValue = true;
+                child.onNext(t);
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                child.onError(e);
+            }
+
+            @Override
+            public void onCompleted() {
+                if (!hasValue) {
+                    try {
+                        child.onNext(defaultValue);
+                    } catch (Throwable e) {
+                        child.onError(e);
+                        return;
+                    }
+                }
+                child.onCompleted();
+            }
+            
+        };
+    }
+    
+}

rxjava-core/src/main/java/rx/operators/OperatorDefaultIfEmpty.java

@@ -0,0 +1,83 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import static org.mockito.Mockito.*;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.Observer;
+import rx.Subscriber;
+
+public class OperatorDefaultIfEmptyTest {
+
+    @Test
+    public void testDefaultIfEmpty() {
+        Observable<Integer> source = Observable.from(1, 2, 3);
+        Observable<Integer> observable = source.defaultIfEmpty(10);
+
+        @SuppressWarnings("unchecked")
+        Observer<Integer> observer = mock(Observer.class);
+        observable.subscribe(observer);
+        verify(observer, never()).onNext(10);
+        verify(observer).onNext(1);
+        verify(observer).onNext(2);
+        verify(observer).onNext(3);
+        verify(observer).onCompleted();
+        verify(observer, never()).onError(any(Throwable.class));
+    }
+
+    @Test
+    public void testDefaultIfEmptyWithEmpty() {
+        Observable<Integer> source = Observable.empty();
+        Observable<Integer> observable = source.defaultIfEmpty(10);
+
+        @SuppressWarnings("unchecked")
+        Observer<Integer> observer = mock(Observer.class);
+        observable.subscribe(observer);
+        
+        verify(observer).onNext(10);
+        verify(observer).onCompleted();
+        verify(observer, never()).onError(any(Throwable.class));
+    }
+    
+    @Test
+    public void testEmptyButClientThrows() {
+        final Observer<Integer> o = mock(Observer.class);
+        
+        Observable.<Integer>empty().defaultIfEmpty(1).subscribe(new Subscriber<Integer>() {
+            @Override
+            public void onNext(Integer t) {
+                throw new OperationReduceTest.CustomException();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                o.onError(e);
+            }
+
+            @Override
+            public void onCompleted() {
+                o.onCompleted();
+            }
+        });
+        
+        verify(o).onError(any(OperationReduceTest.CustomException.class));
+        verify(o, never()).onNext(any(Integer.class));
+        verify(o, never()).onCompleted();
+    }
+}