Implemented the 'Contains' operator

Author: zsxwing

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

@@ -3151,6 +3151,22 @@ public Observable<Boolean> exists(Func1<? super T, Boolean> predicate) {
         return create(OperationAny.exists(this, predicate));
     }
 
+    /**
+     * Determines whether an observable sequence contains a specified element.
+     *
+     * @param value
+     *            The element to search in the sequence.
+     * @return an Observable that emits if the element is in the source sequence.
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh228965(v=vs.103).aspx">MSDN: Observable.Contains</a>
+     */
+    public Observable<Boolean> contains(final T element) {
+        return exists(new Func1<T, Boolean>() {
+            public Boolean call(T t1) {
+                return element == null ? t1 == null : element.equals(t1);
+            }
+        });
+    }
+
     /**
      * Registers an {@link Action0} to be called when this Observable invokes {@link Observer#onCompleted onCompleted} or {@link Observer#onError onError}.
      * <p>

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

@@ -741,4 +741,45 @@ public void testOfTypeWithPolymorphism() {
         verify(aObserver, times(1)).onCompleted();
     }
 
+    @Test
+    public void testContains() {
+        Observable<Boolean> observable = Observable.from("a", "b", null).contains("b");
+
+        @SuppressWarnings("unchecked")
+        Observer<Object> aObserver = mock(Observer.class);
+        observable.subscribe(aObserver);
+        verify(aObserver, times(1)).onNext(true);
+        verify(aObserver, never()).onNext(false);
+        verify(aObserver, never()).onError(
+                org.mockito.Matchers.any(Throwable.class));
+        verify(aObserver, times(1)).onCompleted();
+    }
+
+    @Test
+    public void testContainsWithNull() {
+        Observable<Boolean> observable = Observable.from("a", "b", null).contains(null);
+
+        @SuppressWarnings("unchecked")
+        Observer<Object> aObserver = mock(Observer.class);
+        observable.subscribe(aObserver);
+        verify(aObserver, times(1)).onNext(true);
+        verify(aObserver, never()).onNext(false);
+        verify(aObserver, never()).onError(
+                org.mockito.Matchers.any(Throwable.class));
+        verify(aObserver, times(1)).onCompleted();
+    }
+
+    @Test
+    public void testContainsWithEmptyObservable() {
+        Observable<Boolean> observable = Observable.<String>empty().contains("a");
+
+        @SuppressWarnings("unchecked")
+        Observer<Object> aObserver = mock(Observer.class);
+        observable.subscribe(aObserver);
+        verify(aObserver, times(1)).onNext(false);
+        verify(aObserver, never()).onNext(true);
+        verify(aObserver, never()).onError(
+                org.mockito.Matchers.any(Throwable.class));
+        verify(aObserver, times(1)).onCompleted();
+    }
 }