Fixed the blocking/non-blocking first

Author: zsxwing

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

@@ -5093,35 +5093,33 @@ public Observable<T> skip(int num) {
 
     /**
      * Returns an Observable that emits only the very first item emitted by the
-     * source Observable.
+     * source Observable, or an <code>IllegalArgumentException</code> if the source
+     * {@link Observable} is empty.
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/first.png">
      * 
-     * @return an Observable that emits only the very first item emitted by the
-     *         source Observable, or nothing if the source Observable completes
-     *         without emitting a single item
+     * @return an Observable that emits only the very first item from the
+     *         source, or an <code>IllegalArgumentException</code> if the source {@link Observable} is empty.
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#first">RxJava Wiki: first()</a>
-     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229177.aspx">MSDN: Observable.First</a>
      */
     public Observable<T> first() {
-        return take(1);
+        return takeFirst().last();
     }
 
     /**
      * Returns an Observable that emits only the very first item emitted by the
-     * source Observable that satisfies a given condition.
+     * source Observable that satisfies a given condition, or an <code>IllegalArgumentException</code>
+     * if no such items are emitted.
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/firstN.png">
      * 
      * @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 nothing if the source
-     *         Observable completes without emitting a single matching item
+     *         given condition from the source, or an <code>IllegalArgumentException</code> if no such items are emitted.
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#first">RxJava Wiki: first()</a>
-     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229177.aspx">MSDN: Observable.First</a>
      */
     public Observable<T> first(Func1<? super T, Boolean> predicate) {
-        return skipWhile(not(predicate)).take(1);
+        return takeFirst(predicate).last();
     }
 
     /**
@@ -5245,14 +5243,13 @@ public Observable<T> takeWhileWithIndex(final Func2<? super T, ? super Integer,
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/first.png">
      * 
      * @return an Observable that emits only the very first item from the
-     *         source, or none if the source Observable completes without
+     *         source, or an empty Observable if the source Observable completes without
      *         emitting a single item
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#first">RxJava Wiki: first()</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229177.aspx">MSDN: Observable.First</a>
-     * @see #first()
      */
     public Observable<T> takeFirst() {
-        return first();
+        return take(1);
     }
 
     /**
@@ -5263,14 +5260,13 @@ public Observable<T> takeFirst() {
      * 
      * @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
+     *         given condition from the source, or an empty Observable if the source Observable
      *         completes without emitting a single matching item
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#first">RxJava Wiki: first()</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229177.aspx">MSDN: Observable.First</a>
-     * @see #first(Func1)
      */
     public Observable<T> takeFirst(Func1<? super T, Boolean> predicate) {
-        return first(predicate);
+        return skipWhile(not(predicate)).take(1);
     }
 
     /**

rxjava-core/src/main/java/rx/observables/BlockingObservable.java

@@ -172,6 +172,75 @@ public Iterator<T> getIterator() {
         return OperationToIterator.toIterator(o);
     }
 
+    /**
+     * Returns the first item emitted by a specified {@link Observable},
+     * or IllegalArgumentException if source contains no elements
+     * 
+     * @return the first item emitted by the source {@link Observable}
+     * @throws IllegalArgumentException
+     *            if source contains no elements
+     */
+    public T first() {
+        return from(o.first()).single();
+    }
+
+    /**
+     * Returns the first item emitted by a specified {@link Observable} that matches a predicate,
+     * or IllegalArgumentException if no such items are emitted.
+     * 
+     * @param predicate
+     *            a predicate function to evaluate items emitted by the {@link Observable}
+     * @return the first item emitted by the {@link Observable} that matches the predicate
+     * @throws IllegalArgumentException
+     *            if no such items are emitted.
+     */
+    public T first(Func1<? super T, Boolean> predicate) {
+        return from(o.first(predicate)).single();
+    }
+
+    /**
+     * Returns the first item emitted by a specified {@link Observable}, or a default value if no
+     * items are emitted.
+     * 
+     * @param defaultValue
+     *            a default value to return if the {@link Observable} emits no items
+     * @return the first item emitted by the {@link Observable}, or the default value if no items
+     *         are emitted
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229320(v=vs.103).aspx">MSDN: Observable.FirstOrDefault</a>
+     */
+    public T firstOrDefault(T defaultValue) {
+        boolean found = false;
+        T result = null;
+
+        for (T value : toIterable()) {
+            found = true;
+            result = value;
+            break;
+        }
+
+        if (!found) {
+            return defaultValue;
+        }
+
+        return result;
+    }
+
+    /**
+     * Returns the first item emitted by a specified {@link Observable} that matches a predicate, or
+     * a default value if no such items are emitted.
+     * 
+     * @param defaultValue
+     *            a default value to return if the {@link Observable} emits no matching items
+     * @param predicate
+     *            a predicate function to evaluate items emitted by the {@link Observable}
+     * @return the first item emitted by the {@link Observable} that matches the predicate, or the
+     *         default value if no matching items are emitted
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229759(v=vs.103).aspx">MSDN: Observable.FirstOrDefault</a>
+     */
+    public T firstOrDefault(T defaultValue, Func1<? super T, Boolean> predicate) {
+        return from(o.filter(predicate)).firstOrDefault(defaultValue);
+    }
+
     /**
      * Returns the last item emitted by a specified {@link Observable}.
      * <p>

rxjava-core/src/test/java/rx/ObservableTests.java

@@ -164,43 +164,61 @@ public Subscription onSubscribe(Observer<? super String> obsv) {
         verify(w, times(1)).onError(any(RuntimeException.class));
     }
 
-    public void testFirstWithPredicateOfSome() {
+    public void testTakeFirstWithPredicateOfSome() {
         Observable<Integer> observable = Observable.from(1, 3, 5, 4, 6, 3);
-        observable.first(IS_EVEN).subscribe(w);
+        observable.takeFirst(IS_EVEN).subscribe(w);
         verify(w, times(1)).onNext(anyInt());
         verify(w).onNext(4);
         verify(w, times(1)).onCompleted();
         verify(w, never()).onError(any(Throwable.class));
     }
 
     @Test
-    public void testFirstWithPredicateOfNoneMatchingThePredicate() {
+    public void testTakeFirstWithPredicateOfNoneMatchingThePredicate() {
         Observable<Integer> observable = Observable.from(1, 3, 5, 7, 9, 7, 5, 3, 1);
-        observable.first(IS_EVEN).subscribe(w);
+        observable.takeFirst(IS_EVEN).subscribe(w);
         verify(w, never()).onNext(anyInt());
         verify(w, times(1)).onCompleted();
         verify(w, never()).onError(any(Throwable.class));
     }
 
     @Test
-    public void testFirstOfSome() {
+    public void testTakeFirstOfSome() {
         Observable<Integer> observable = Observable.from(1, 2, 3);
-        observable.first().subscribe(w);
+        observable.takeFirst().subscribe(w);
         verify(w, times(1)).onNext(anyInt());
         verify(w).onNext(1);
         verify(w, times(1)).onCompleted();
         verify(w, never()).onError(any(Throwable.class));
     }
 
     @Test
-    public void testFirstOfNone() {
+    public void testTakeFirstOfNone() {
         Observable<Integer> observable = Observable.empty();
-        observable.first().subscribe(w);
+        observable.takeFirst().subscribe(w);
         verify(w, never()).onNext(anyInt());
         verify(w, times(1)).onCompleted();
         verify(w, never()).onError(any(Throwable.class));
     }
 
+    @Test
+    public void testFirstOfNone() {
+        Observable<Integer> observable = Observable.empty();
+        observable.first().subscribe(w);
+        verify(w, never()).onNext(anyInt());
+        verify(w, never()).onCompleted();
+        verify(w, times(1)).onError(isA(IllegalArgumentException.class));
+    }
+
+    @Test
+    public void testFirstWithPredicateOfNoneMatchingThePredicate() {
+        Observable<Integer> observable = Observable.from(1, 3, 5, 7, 9, 7, 5, 3, 1);
+        observable.first(IS_EVEN).subscribe(w);
+        verify(w, never()).onNext(anyInt());
+        verify(w, never()).onCompleted();
+        verify(w, times(1)).onError(isA(IllegalArgumentException.class));
+    }
+
     @Test
     public void testReduce() {
         Observable<Integer> observable = Observable.from(1, 2, 3, 4);

rxjava-core/src/test/java/rx/observables/BlockingObservableTest.java

@@ -257,6 +257,76 @@ public void call(String t1) {
         }
     }
 
+    @Test
+    public void testFirst() {
+        BlockingObservable<String> observable = BlockingObservable.from(Observable.from("one", "two", "three"));
+        assertEquals("one", observable.first());
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testFirstWithEmpty() {
+        BlockingObservable.from(Observable.<String>empty()).first();
+    }
+
+    @Test
+    public void testFirstWithPredicate() {
+        BlockingObservable<String> observable = BlockingObservable.from(Observable.from("one", "two", "three"));
+        String first = observable.first(new Func1<String, Boolean>() {
+            @Override
+            public Boolean call(String args) {
+                return args.length() > 3;
+            }
+        });
+        assertEquals("three", first);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testFirstWithPredicateAndEmpty() {
+        BlockingObservable<String> observable = BlockingObservable.from(Observable.from("one", "two", "three"));
+        observable.first(new Func1<String, Boolean>() {
+            @Override
+            public Boolean call(String args) {
+                return args.length() > 5;
+            }
+        });
+    }
+
+    @Test
+    public void testFirstOrDefault() {
+        BlockingObservable<String> observable = BlockingObservable.from(Observable.from("one", "two", "three"));
+        assertEquals("one", observable.firstOrDefault("default"));
+    }
+
+    @Test
+    public void testFirstOrDefaultWithEmpty() {
+        BlockingObservable<String> observable = BlockingObservable.from(Observable.<String>empty());
+        assertEquals("default", observable.firstOrDefault("default"));
+    }
+
+    @Test
+    public void testFirstOrDefaultWithPredicate() {
+        BlockingObservable<String> observable = BlockingObservable.from(Observable.from("one", "two", "three"));
+        String first = observable.firstOrDefault("default", new Func1<String, Boolean>() {
+            @Override
+            public Boolean call(String args) {
+                return args.length() > 3;
+            }
+        });
+        assertEquals("three", first);
+    }
+
+    @Test
+    public void testFirstOrDefaultWithPredicateAndEmpty() {
+        BlockingObservable<String> observable = BlockingObservable.from(Observable.from("one", "two", "three"));
+        String first = observable.firstOrDefault("default", new Func1<String, Boolean>() {
+            @Override
+            public Boolean call(String args) {
+                return args.length() > 5;
+            }
+        });
+        assertEquals("default", first);
+    }
+
     private static class TestException extends RuntimeException {
         private static final long serialVersionUID = 1L;
     }