Added the two variants of the first operator.

Author: jmhofer

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;
@@ -2275,6 +2277,26 @@ 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.
+     * @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(Func1<? super T, Boolean> predicate) {
+        return skipWhile(not(predicate)).take(1);
+    }
+    
     /**
      * Returns an Observable that emits items emitted by the source Observable so long as a
      * specified condition is true.

rxjava-core/src/main/java/rx/util/functions/Functions.java

@@ -313,6 +313,16 @@ public Void call(Object... args) {
         };
     }
 
+    /**
+     * Constructs a predicate that returns true for each input that the source
+     * predicate returns false for and vice versa.
+     * 
+     * @param predicate The source predicate to negate.
+     */
+    public static <T> Func1<T, Boolean> not(Func1<? super T, Boolean> predicate) {
+        return new Not<T>(predicate);
+    }
+
     public static <T> Func1<? super T, Boolean> alwaysTrue() {
         return AlwaysTrue.INSTANCE;
     }
@@ -334,4 +344,5 @@ public Boolean call(Object o) {
             return true;
         }
     }
+    
 }

rxjava-core/src/main/java/rx/util/functions/Not.java

@@ -0,0 +1,40 @@
+/**
+ * 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.util.functions;
+
+/**
+ * Implements the negation of a predicate. 
+ * 
+ * @param <T> The type of the single input parameter.
+ */
+public class Not<T> implements Func1<T, Boolean> {
+    private final Func1<? super T, Boolean> predicate;
+ 
+    /**
+     * Constructs a predicate that returns true for each input that the source
+     * predicate returns false for and vice versa.
+     * 
+     * @param predicate The source predicate to negate.
+     */
+    public Not(Func1<? super T, Boolean> predicate) {
+        this.predicate = predicate;
+    }
+    
+    @Override
+    public Boolean call(T param) {
+        return !predicate.call(param);
+    }
+}

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

@@ -35,13 +35,21 @@
 import rx.subscriptions.BooleanSubscription;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action1;
+import rx.util.functions.Func1;
 import rx.util.functions.Func2;
 
 public class ObservableTests {
 
     @Mock
     Observer<Integer> w;
 
+    private static final Func1<Integer, Boolean> IS_EVEN = new Func1<Integer, Boolean>() {
+        @Override
+        public Boolean call(Integer value) {
+            return value % 2 == 0;
+        }
+    };
+    
     @Before
     public void before() {
         MockitoAnnotations.initMocks(this);
@@ -73,6 +81,44 @@ public Subscription onSubscribe(Observer<? super String> Observer) {
         verify(aObserver, times(1)).onCompleted();
     }
 
+    @Test
+    public void testFirstWithPredicateOfSome() {
+        Observable<Integer> observable = Observable.from(1, 3, 5, 4, 6, 3);
+        observable.first(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() {
+        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, times(1)).onCompleted();
+        verify(w, never()).onError(any(Throwable.class));
+    }
+    
+    @Test
+    public void testFirstOfSome() {
+        Observable<Integer> observable = Observable.from(1, 2, 3);
+        observable.first().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() {
+        Observable<Integer> observable = Observable.empty();
+        observable.first().subscribe(w);
+        verify(w, never()).onNext(anyInt());
+        verify(w, times(1)).onCompleted();
+        verify(w, never()).onError(any(Throwable.class));
+    }
+    
     @Test
     public void testReduce() {
         Observable<Integer> observable = Observable.from(1, 2, 3, 4);