Implemented "First" and "FirstOrDefault" operations

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

@@ -15,6 +15,8 @@
  */
 package rx;
 
+import static rx.util.functions.Functions.not;
+
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -35,6 +37,7 @@
 import rx.operators.OperationDematerialize;
 import rx.operators.OperationFilter;
 import rx.operators.OperationFinally;
+import rx.operators.OperationFirstOrDefault;
 import rx.operators.OperationGroupBy;
 import rx.operators.OperationMap;
 import rx.operators.OperationMaterialize;
@@ -49,6 +52,7 @@
 import rx.operators.OperationSample;
 import rx.operators.OperationScan;
 import rx.operators.OperationSkip;
+import rx.operators.OperationSkipWhile;
 import rx.operators.OperationSubscribeOn;
 import rx.operators.OperationSwitch;
 import rx.operators.OperationSynchronize;
@@ -2274,6 +2278,58 @@ public Observable<T> take(final int num) {
         return create(OperationTake.take(this, num));
     }
 
+    /**
+     * Returns an Observable that emits only the very first item emitted by the source Observable.
+     * @return an Observable that emits only the very first item from the source, or none if the
+     *         source Observable completes without emitting a single item.
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229177%28v=vs.103%29.aspx">MSDN: Observable.First</a>
+     */
+    public Observable<T> first() {
+        return take(1);
+    }
+
+    /**
+     * Returns an Observable that emits only the very first item emitted by the source Observable
+     * that satisfies a given condition.
+     * @param predicate
+     *            The condition any source emitted item has to satisfy.
+     * @return an Observable that emits only the very first item satisfying the given condition from the source, 
+     *         or none if the source Observable completes without emitting a single matching item.
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229177%28v=vs.103%29.aspx">MSDN: Observable.First</a>
+     */
+    public Observable<T> first(Func1<? super T, Boolean> predicate) {
+        return skipWhile(not(predicate)).take(1);
+    }
+
+    /**
+     * Returns an Observable that emits only the very first item emitted by the source Observable, or
+     * a default value.
+     * @param defaultValue
+     *            The default value to emit if the source Observable doesn't emit anything.
+     * @return an Observable that emits only the very first item from the source, or a default value 
+     *         if the source Observable completes without emitting a single item.
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229320%28v=vs.103%29.aspx">MSDN: Observable.FirstOrDefault</a>
+     */
+    public Observable<T> firstOrDefault(T defaultValue) {
+        return create(OperationFirstOrDefault.firstOrDefault(this, defaultValue));
+    }
+
+    /**
+     * Returns an Observable that emits only the very first item emitted by the source Observable
+     * that satisfies a given condition, or a default value otherwise.
+     * @param predicate
+     *            The condition any source emitted item has to satisfy.
+     * @param defaultValue
+     *            The default value to emit if the source Observable doesn't emit anything that 
+     *            satisfies the given condition.
+     * @return an Observable that emits only the very first item from the source that satisfies the
+     *         given condition, or a default value otherwise.
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229759%28v=vs.103%29.aspx">MSDN: Observable.FirstOrDefault</a>
+     */
+    public Observable<T> firstOrDefault(Func1<? super T, Boolean> predicate, T defaultValue) {
+        return create(OperationFirstOrDefault.firstOrDefault(this, predicate, defaultValue));
+    }
+
     /**
      * Returns an Observable that emits items emitted by the source Observable so long as a
      * specified condition is true.
@@ -2341,6 +2397,34 @@ public <E> Observable<T> takeUntil(Observable<? extends E> other) {
         return OperationTakeUntil.takeUntil(this, other);
     }
 
+    /**
+     * Returns an Observable that bypasses all items from the source Observable as long as the specified
+     * condition holds true. Emits all further source items as soon as the condition becomes false.
+     * @param predicate
+     *            A function to test each item emitted from the source Observable for a condition.
+     *            It receives the emitted item as first parameter and the index of the emitted item as
+     *            second parameter.
+     * @return an Observable that emits all items from the source Observable as soon as the condition
+     *         becomes false. 
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh211631%28v=vs.103%29.aspx">MSDN: Observable.SkipWhile</a>
+     */
+    public Observable<T> skipWhileWithIndex(Func2<? super T, Integer, Boolean> predicate) {
+        return create(OperationSkipWhile.skipWhileWithIndex(this, predicate));
+    }
+
+    /**
+     * Returns an Observable that bypasses all items from the source Observable as long as the specified
+     * condition holds true. Emits all further source items as soon as the condition becomes false.
+     * @param predicate
+     *            A function to test each item emitted from the source Observable for a condition.
+     * @return an Observable that emits all items from the source Observable as soon as the condition
+     *         becomes false. 
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229685%28v=vs.103%29.aspx">MSDN: Observable.SkipWhile</a>
+     */
+    public Observable<T> skipWhile(Func1<? super T, Boolean> predicate) {
+        return create(OperationSkipWhile.skipWhile(this, predicate));
+    }
+
     /**
      * Returns an Observable that emits a single item, a list composed of all the items emitted by
      * the source Observable.

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

@@ -0,0 +1,190 @@
+/**
+ * Copyright 2013 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.Matchers.*;
+import static org.mockito.Mockito.*;
+import static org.mockito.MockitoAnnotations.initMocks;
+import static rx.Observable.create;
+import static rx.Observable.empty;
+import static rx.Observable.from;
+import static rx.util.functions.Functions.alwaysTrue;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+
+import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observer;
+import rx.Subscription;
+import rx.subscriptions.Subscriptions;
+import rx.util.functions.Action0;
+import rx.util.functions.Func1;
+
+/**
+ * Returns an Observable that emits the first item emitted by the source
+ * Observable, or a default value if the source emits nothing.
+ */
+public final class OperationFirstOrDefault {
+
+    /**
+     * Returns an Observable that emits the first item emitted by the source
+     * Observable that satisfies the given condition, 
+     * or a default value if the source emits no items that satisfy the given condition.
+     * 
+     * @param source
+     *            The source Observable to emit the first item for.
+     * @param predicate
+     *            The condition the emitted source items have to satisfy.
+     * @param defaultValue
+     *            The default value to use whenever the source Observable doesn't emit anything.
+     * @return A subscription function for creating the target Observable.
+     */
+    public static <T> OnSubscribeFunc<T> firstOrDefault(Observable<? extends T> source, Func1<? super T, Boolean> predicate, T defaultValue) {
+        return new FirstOrElse<T>(source, predicate, defaultValue);
+    }
+
+    /**
+     * Returns an Observable that emits the first item emitted by the source
+     * Observable, or a default value if the source emits nothing.
+     * 
+     * @param source
+     *            The source Observable to emit the first item for.
+     * @param defaultValue
+     *            The default value to use whenever the source Observable doesn't emit anything.
+     * @return A subscription function for creating the target Observable.
+     */
+    public static <T> OnSubscribeFunc<T> firstOrDefault(Observable<? extends T> source, T defaultValue) {
+        return new FirstOrElse<T>(source, alwaysTrue(), defaultValue);
+    }
+
+    private static class FirstOrElse<T> implements OnSubscribeFunc<T> {
+        private final Observable<? extends T> source;
+        private final Func1<? super T, Boolean> predicate;
+        private final T defaultValue;
+
+        private FirstOrElse(Observable<? extends T> source, Func1<? super T, Boolean> predicate, T defaultValue) {
+            this.source = source;
+            this.defaultValue = defaultValue;
+            this.predicate = predicate;
+        }
+
+        @Override
+        public Subscription onSubscribe(final Observer<? super T> observer) {
+            final Subscription sourceSub = source.subscribe(new Observer<T>() {
+                private final AtomicBoolean hasEmitted = new AtomicBoolean(false);
+
+                @Override
+                public void onCompleted() {
+                    if (!hasEmitted.get()) {
+                        observer.onNext(defaultValue);
+                        observer.onCompleted();
+                    }
+                }
+
+                @Override
+                public void onError(Throwable e) {
+                    observer.onError(e);
+                }
+
+                @Override
+                public void onNext(T next) {
+                    try {
+                        if (!hasEmitted.get() && predicate.call(next)) {
+                            hasEmitted.set(true);
+                            observer.onNext(next);
+                            observer.onCompleted();
+                        }
+                    } catch (Throwable t) {
+                        // may happen within the predicate call (user code)
+                        observer.onError(t);
+                    }
+                }
+            });
+
+            return Subscriptions.create(new Action0() {
+                @Override
+                public void call() {
+                    sourceSub.unsubscribe();
+                }
+            });
+        }
+    }
+
+    public static class UnitTest {
+        @Mock
+        Observer<? super String> w;
+
+        private static final Func1<String, Boolean> IS_D = new Func1<String, Boolean>() {
+            @Override
+            public Boolean call(String value) {
+                return "d".equals(value);
+            }
+        };
+
+        @Before
+        public void before() {
+            initMocks(this);
+        }
+
+        @Test
+        public void testFirstOrElseOfNone() {
+            Observable<String> src = empty();
+            create(firstOrDefault(src, "default")).subscribe(w);
+
+            verify(w, times(1)).onNext(anyString());
+            verify(w, times(1)).onNext("default");
+            verify(w, never()).onError(any(Throwable.class));
+            verify(w, times(1)).onCompleted();
+        }
+
+        @Test
+        public void testFirstOrElseOfSome() {
+            Observable<String> src = from("a", "b", "c");
+            create(firstOrDefault(src, "default")).subscribe(w);
+
+            verify(w, times(1)).onNext(anyString());
+            verify(w, times(1)).onNext("a");
+            verify(w, never()).onError(any(Throwable.class));
+            verify(w, times(1)).onCompleted();
+        }
+
+        @Test
+        public void testFirstOrElseWithPredicateOfNoneMatchingThePredicate() {
+            Observable<String> src = from("a", "b", "c");
+            create(firstOrDefault(src, IS_D, "default")).subscribe(w);
+
+            verify(w, times(1)).onNext(anyString());
+            verify(w, times(1)).onNext("default");
+            verify(w, never()).onError(any(Throwable.class));
+            verify(w, times(1)).onCompleted();
+        }
+
+        @Test
+        public void testFirstOrElseWithPredicateOfSome() {
+            Observable<String> src = from("a", "b", "c", "d", "e", "f");
+            create(firstOrDefault(src, IS_D, "default")).subscribe(w);
+
+            verify(w, times(1)).onNext(anyString());
+            verify(w, times(1)).onNext("d");
+            verify(w, never()).onError(any(Throwable.class));
+            verify(w, times(1)).onCompleted();
+        }
+    }
+}