Implemented the 'any' operator

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

@@ -77,6 +77,7 @@
 import rx.operators.OperationZip;
 import rx.operators.SafeObservableSubscription;
 import rx.operators.SafeObserver;
+import rx.operators.OperationAny;
 import rx.plugins.RxJavaErrorHandler;
 import rx.plugins.RxJavaObservableExecutionHook;
 import rx.plugins.RxJavaPlugins;
@@ -4187,4 +4188,35 @@ private boolean isInternalImplementation(Object o) {
         return p != null && p.getName().startsWith("rx.operators");
     }
 
+    /**
+     * Returns an {@link Observable} that emits <code>true</code> if the source
+     * {@link Observable} is not empty, otherwise <code>false</code>.
+     * 
+     * @return A subscription function for creating the target Observable.
+     * @see <a href=
+     *      "http://msdn.microsoft.com/en-us/library/hh229905(v=vs.103).aspx"
+     *      >MSDN: Observable.Any</a>
+     */
+    public Observable<Boolean> any() {
+        return create(OperationAny.any(this));
+    }
+
+    /**
+     * Returns an {@link Observable} that emits <code>true</code> if any element
+     * of the source {@link Observable} satisfies the given condition, otherwise
+     * <code>false</code>. Note: always emit <code>false</code> if the source
+     * {@link Observable} is empty.
+     * 
+     * @param predicate
+     *            The condition to test every element.
+     * @return A subscription function for creating the target Observable.
+     * @see <a href=
+     *      "http://msdn.microsoft.com/en-us/library/hh211993(v=vs.103).aspx"
+     *      >MSDN: Observable.Any</a> Note: the description in this page is
+     *      wrong.
+     */
+    public Observable<Boolean> any(Func1<? super T, Boolean> predicate) {
+        return create(OperationAny.any(this, predicate));
+    }
+
 }

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

@@ -0,0 +1,230 @@
+package rx.operators;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static rx.util.functions.Functions.alwaysTrue;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observer;
+import rx.Subscription;
+import rx.util.functions.Func1;
+
+/**
+ * Returns an {@link Observable} that emits <code>true</code> if any element of
+ * an observable sequence satisfies a condition, otherwise <code>false</code>.
+ */
+public final class OperationAny {
+
+    /**
+     * Returns an {@link Observable} that emits <code>true</code> if the source
+     * {@link Observable} is not empty, otherwise <code>false</code>.
+     * 
+     * @param source
+     *            The source {@link Observable} to check if not empty.
+     * @return A subscription function for creating the target Observable.
+     */
+    public static <T> OnSubscribeFunc<Boolean> any(
+            Observable<? extends T> source) {
+        return new Any<T>(source, alwaysTrue());
+    }
+
+    /**
+     * Returns an {@link Observable} that emits <code>true</code> if any element
+     * of the source {@link Observable} satisfies the given condition, otherwise
+     * <code>false</code>. Note: always emit <code>false</code> if the source
+     * {@link Observable} is empty.
+     * 
+     * @param source
+     *            The source {@link Observable} to check if any element
+     *            satisfies the given condition.
+     * @param predicate
+     *            The condition to test every element.
+     * @return A subscription function for creating the target Observable.
+     */
+    public static <T> OnSubscribeFunc<Boolean> any(
+            Observable<? extends T> source, Func1<? super T, Boolean> predicate) {
+        return new Any<T>(source, predicate);
+    }
+
+    private static class Any<T> implements OnSubscribeFunc<Boolean> {
+
+        private final Observable<? extends T> source;
+        private final Func1<? super T, Boolean> predicate;
+
+        private Any(Observable<? extends T> source,
+                Func1<? super T, Boolean> predicate) {
+            this.source = source;
+            this.predicate = predicate;
+        }
+
+        @Override
+        public Subscription onSubscribe(final Observer<? super Boolean> observer) {
+            final SafeObservableSubscription subscription = new SafeObservableSubscription();
+            return subscription.wrap(source.subscribe(new Observer<T>() {
+
+                private final AtomicBoolean hasEmitted = new AtomicBoolean(
+                        false);
+
+                @Override
+                public void onNext(T value) {
+                    try {
+                        if (hasEmitted.get() == false) {
+                            if (predicate.call(value) == true
+                                    && hasEmitted.getAndSet(true) == false) {
+                                observer.onNext(true);
+                                observer.onCompleted();
+                                // this will work if the sequence is
+                                // asynchronous, it
+                                // will have no effect on a synchronous
+                                // observable
+                                subscription.unsubscribe();
+                            }
+                        }
+                    } catch (Throwable ex) {
+                        observer.onError(ex);
+                        // this will work if the sequence is asynchronous, it
+                        // will have no effect on a synchronous observable
+                        subscription.unsubscribe();
+                    }
+
+                }
+
+                @Override
+                public void onError(Throwable ex) {
+                    observer.onError(ex);
+                }
+
+                @Override
+                public void onCompleted() {
+                    if (!hasEmitted.get()) {
+                        observer.onNext(false);
+                        observer.onCompleted();
+                    }
+                }
+            }));
+        }
+
+    }
+
+    public static class UnitTest {
+
+        @Test
+        public void testAnyWithTwoItems() {
+            Observable<Integer> w = Observable.from(1, 2);
+            Observable<Boolean> observable = Observable.create(any(w));
+
+            @SuppressWarnings("unchecked")
+            Observer<Boolean> aObserver = mock(Observer.class);
+            observable.subscribe(aObserver);
+            verify(aObserver, never()).onNext(false);
+            verify(aObserver, times(1)).onNext(true);
+            verify(aObserver, never()).onError(
+                    org.mockito.Matchers.any(Throwable.class));
+            verify(aObserver, times(1)).onCompleted();
+        }
+
+        @Test
+        public void testAnyWithOneItem() {
+            Observable<Integer> w = Observable.from(1);
+            Observable<Boolean> observable = Observable.create(any(w));
+
+            @SuppressWarnings("unchecked")
+            Observer<Boolean> aObserver = mock(Observer.class);
+            observable.subscribe(aObserver);
+            verify(aObserver, never()).onNext(false);
+            verify(aObserver, times(1)).onNext(true);
+            verify(aObserver, never()).onError(
+                    org.mockito.Matchers.any(Throwable.class));
+            verify(aObserver, times(1)).onCompleted();
+        }
+
+        @Test
+        public void testAnyWithEmpty() {
+            Observable<Integer> w = Observable.empty();
+            Observable<Boolean> observable = Observable.create(any(w));
+
+            @SuppressWarnings("unchecked")
+            Observer<Boolean> 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();
+        }
+
+        @Test
+        public void testAnyWithPredicate1() {
+            Observable<Integer> w = Observable.from(1, 2, 3);
+            Observable<Boolean> observable = Observable.create(any(w,
+                    new Func1<Integer, Boolean>() {
+
+                        @Override
+                        public Boolean call(Integer t1) {
+                            return t1 < 2;
+                        }
+                    }));
+
+            @SuppressWarnings("unchecked")
+            Observer<Boolean> aObserver = mock(Observer.class);
+            observable.subscribe(aObserver);
+            verify(aObserver, never()).onNext(false);
+            verify(aObserver, times(1)).onNext(true);
+            verify(aObserver, never()).onError(
+                    org.mockito.Matchers.any(Throwable.class));
+            verify(aObserver, times(1)).onCompleted();
+        }
+
+        @Test
+        public void testAnyWithPredicate2() {
+            Observable<Integer> w = Observable.from(1, 2, 3);
+            Observable<Boolean> observable = Observable.create(any(w,
+                    new Func1<Integer, Boolean>() {
+
+                        @Override
+                        public Boolean call(Integer t1) {
+                            return t1 < 1;
+                        }
+                    }));
+
+            @SuppressWarnings("unchecked")
+            Observer<Boolean> 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();
+        }
+
+        @Test
+        public void testAnyWithEmptyAndPredicate() {
+            // If the source is empty, always output false.
+            Observable<Integer> w = Observable.empty();
+            Observable<Boolean> observable = Observable.create(any(w,
+                    new Func1<Integer, Boolean>() {
+
+                        @Override
+                        public Boolean call(Integer t1) {
+                            return true;
+                        }
+                    }));
+
+            @SuppressWarnings("unchecked")
+            Observer<Boolean> 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();
+        }
+    }
+}

rxjava-core/src/main/java/rx/subjects/AsyncSubject.java

@@ -168,7 +168,7 @@ public void testCompleted() {
         private void assertCompletedObserver(Observer<String> aObserver)
         {
             verify(aObserver, times(1)).onNext("three");
-            verify(aObserver, Mockito.never()).onError(any(Throwable.class));
+            verify(aObserver, Mockito.never()).onError(org.mockito.Matchers.any(Throwable.class));
             verify(aObserver, times(1)).onCompleted();
         }
 
@@ -221,7 +221,7 @@ public void testUnsubscribeBeforeCompleted() {
         private void assertNoOnNextEventsReceived(Observer<String> aObserver)
         {
             verify(aObserver, Mockito.never()).onNext(anyString());
-            verify(aObserver, Mockito.never()).onError(any(Throwable.class));
+            verify(aObserver, Mockito.never()).onError(org.mockito.Matchers.any(Throwable.class));
             verify(aObserver, Mockito.never()).onCompleted();
         }
 

rxjava-core/src/main/java/rx/subjects/BehaviorSubject.java

@@ -15,7 +15,6 @@
  */
 package rx.subjects;
 
-import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 
 import java.util.concurrent.ConcurrentHashMap;
@@ -199,7 +198,7 @@ private void assertCompletedObserver(Observer<String> aObserver)
         {
             verify(aObserver, times(1)).onNext("default");
             verify(aObserver, times(1)).onNext("one");
-            verify(aObserver, Mockito.never()).onError(any(Throwable.class));
+            verify(aObserver, Mockito.never()).onError(org.mockito.Matchers.any(Throwable.class));
             verify(aObserver, times(1)).onCompleted();
         }
 

rxjava-core/src/main/java/rx/subjects/PublishSubject.java

@@ -16,7 +16,6 @@
 package rx.subjects;
 
 import static org.junit.Assert.*;
-import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 
 import java.util.ArrayList;
@@ -273,7 +272,7 @@ private void assertCompletedObserver(Observer<String> aObserver)
             verify(aObserver, times(1)).onNext("one");
             verify(aObserver, times(1)).onNext("two");
             verify(aObserver, times(1)).onNext("three");
-            verify(aObserver, Mockito.never()).onError(any(Throwable.class));
+            verify(aObserver, Mockito.never()).onError(org.mockito.Matchers.any(Throwable.class));
             verify(aObserver, times(1)).onCompleted();
         }
 
@@ -340,7 +339,7 @@ private void assertCompletedStartingWithThreeObserver(Observer<String> aObserver
             verify(aObserver, Mockito.never()).onNext("one");
             verify(aObserver, Mockito.never()).onNext("two");
             verify(aObserver, times(1)).onNext("three");
-            verify(aObserver, Mockito.never()).onError(any(Throwable.class));
+            verify(aObserver, Mockito.never()).onError(org.mockito.Matchers.any(Throwable.class));
             verify(aObserver, times(1)).onCompleted();
         }
 
@@ -374,7 +373,7 @@ private void assertObservedUntilTwo(Observer<String> aObserver)
             verify(aObserver, times(1)).onNext("one");
             verify(aObserver, times(1)).onNext("two");
             verify(aObserver, Mockito.never()).onNext("three");
-            verify(aObserver, Mockito.never()).onError(any(Throwable.class));
+            verify(aObserver, Mockito.never()).onError(org.mockito.Matchers.any(Throwable.class));
             verify(aObserver, Mockito.never()).onCompleted();
         }
 
@@ -404,7 +403,7 @@ public void testUnsubscribeAfterOnCompleted() {
             inOrder.verify(anObserver, times(1)).onNext("one");
             inOrder.verify(anObserver, times(1)).onNext("two");
             inOrder.verify(anObserver, times(1)).onCompleted();
-            inOrder.verify(anObserver, Mockito.never()).onError(any(Throwable.class));
+            inOrder.verify(anObserver, Mockito.never()).onError(org.mockito.Matchers.any(Throwable.class));
 
             @SuppressWarnings("unchecked")
             Observer<String> anotherObserver = mock(Observer.class);
@@ -414,7 +413,7 @@ public void testUnsubscribeAfterOnCompleted() {
             inOrder.verify(anotherObserver, Mockito.never()).onNext("one");
             inOrder.verify(anotherObserver, Mockito.never()).onNext("two");
             inOrder.verify(anotherObserver, times(1)).onCompleted();
-            inOrder.verify(anotherObserver, Mockito.never()).onError(any(Throwable.class));
+            inOrder.verify(anotherObserver, Mockito.never()).onError(org.mockito.Matchers.any(Throwable.class));
         }
 
         @Test

rxjava-core/src/main/java/rx/subjects/ReplaySubject.java

@@ -15,7 +15,6 @@
  */
 package rx.subjects;
 
-import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 
 import java.util.ArrayList;
@@ -215,7 +214,7 @@ private void assertCompletedObserver(Observer<String> aObserver)
             inOrder.verify(aObserver, times(1)).onNext("one");
             inOrder.verify(aObserver, times(1)).onNext("two");
             inOrder.verify(aObserver, times(1)).onNext("three");
-            inOrder.verify(aObserver, Mockito.never()).onError(any(Throwable.class));
+            inOrder.verify(aObserver, Mockito.never()).onError(org.mockito.Matchers.any(Throwable.class));
             inOrder.verify(aObserver, times(1)).onCompleted();
             inOrder.verifyNoMoreInteractions();
         }
@@ -307,7 +306,7 @@ private void assertObservedUntilTwo(Observer<String> aObserver)
             verify(aObserver, times(1)).onNext("one");
             verify(aObserver, times(1)).onNext("two");
             verify(aObserver, Mockito.never()).onNext("three");
-            verify(aObserver, Mockito.never()).onError(any(Throwable.class));
+            verify(aObserver, Mockito.never()).onError(org.mockito.Matchers.any(Throwable.class));
             verify(aObserver, Mockito.never()).onCompleted();
         }