Merge branch 'OperatorAll' of github.com:akarnokd/RxJava into operator-all

Conflicts:
	rxjava-core/src/main/java/rx/operators/OperatorAll.java
	rxjava-core/src/test/java/rx/operators/OperatorAllTest.java

Author: benjchristensen

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

@@ -48,7 +48,7 @@
 import rx.observers.SafeSubscriber;
 import rx.operators.OnSubscribeFromIterable;
 import rx.operators.OnSubscribeRange;
-import rx.operators.OperationAll;
+import rx.operators.OperatorAll;
 import rx.operators.OperationAny;
 import rx.operators.OperationAsObservable;
 import rx.operators.OperationBuffer;

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

@@ -1,22 +1,37 @@
+/**
+ * Copyright 2014 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 java.util.concurrent.atomic.AtomicBoolean;
+
 import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observable.Operator;
+import rx.Observer;
 import rx.Subscriber;
-import rx.exceptions.OnErrorThrowable;
+import rx.Subscription;
 import rx.functions.Func1;
 
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import static rx.Observable.Operator;
-
 /**
  * Returns an Observable that emits a Boolean that indicates whether all items emitted by an
  * Observable satisfy a condition.
  * <p>
  * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/all.png">
  */
-public class OperatorAll<T> implements Operator<Boolean,T>{
-
+public final class OperatorAll<T> implements Operator<Boolean, T> {
     private final Func1<? super T, Boolean> predicate;
 
     public OperatorAll(Func1<? super T, Boolean> predicate) {
@@ -25,13 +40,17 @@ public OperatorAll(Func1<? super T, Boolean> predicate) {
 
     @Override
     public Subscriber<? super T> call(final Subscriber<? super Boolean> child) {
-        return new Subscriber<T>() {
-            private AtomicBoolean status = new AtomicBoolean(true);
-
+        Subscriber s = new Subscriber<T>() {
+            boolean done;
             @Override
-            public void onCompleted() {
-                child.onNext(status.get());
-                child.onCompleted();
+            public void onNext(T t) {
+                boolean result = predicate.call(t);
+                if (!result && !done) {
+                    done = true;
+                    child.onNext(false);
+                    child.onCompleted();
+                    unsubscribe();
+                }
             }
 
             @Override
@@ -40,19 +59,15 @@ public void onError(Throwable e) {
             }
 
             @Override
-            public void onNext(T t) {
-                try {
-                    final Boolean result = predicate.call(t);
-                    boolean changed = status.compareAndSet(true, result);
-
-                    if (changed && !result) {
-                        child.onNext(false);
-                        child.onCompleted();
-                    }
-                } catch (Throwable e) {
-                    child.onError(OnErrorThrowable.addValueAsLastCause(e,t));
+            public void onCompleted() {
+                if (!done) {
+                    done = true;
+                    child.onNext(true);
+                    child.onCompleted();
                 }
             }
         };
+        child.add(s);
+        return s;
     }
 }

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

@@ -19,31 +19,26 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 
-import java.util.Arrays;
-
 import org.junit.Test;
 
 import rx.Observable;
 import rx.Observer;
 import rx.functions.Func1;
-import rx.observers.TestSubscriber;
 
 public class OperatorAllTest {
 
-    final Func1<String, Boolean> hasLength3 = new Func1<String, Boolean>() {
-        @Override
-        public Boolean call(String s) {
-            return s.length() == 3;
-        }
-    };
-
     @Test
     @SuppressWarnings("unchecked")
     public void testAll() {
-        Observable<Boolean> obs = Observable.from(Arrays.asList("one", "two", "six")).all(hasLength3);
+        Observable<String> obs = Observable.from("one", "two", "six");
 
         Observer<Boolean> observer = mock(Observer.class);
-        obs.subscribe(new TestSubscriber<Boolean>(observer));
+        obs.all(new Func1<String, Boolean>() {
+            @Override
+            public Boolean call(String s) {
+                return s.length() == 3;
+            }
+        }).subscribe(observer);
 
         verify(observer).onNext(true);
         verify(observer).onCompleted();
@@ -53,10 +48,15 @@ public void testAll() {
     @Test
     @SuppressWarnings("unchecked")
     public void testNotAll() {
-        Observable<Boolean> obs = Observable.from(Arrays.asList("one", "two", "three", "six")).all(hasLength3);
+        Observable<String> obs = Observable.from("one", "two", "three", "six");
 
         Observer<Boolean> observer = mock(Observer.class);
-        obs.subscribe(new TestSubscriber<Boolean>(observer));
+        obs.all(new Func1<String, Boolean>() {
+            @Override
+            public Boolean call(String s) {
+                return s.length() == 3;
+            }
+        }).subscribe(observer);
 
         verify(observer).onNext(false);
         verify(observer).onCompleted();
@@ -66,10 +66,15 @@ public void testNotAll() {
     @Test
     @SuppressWarnings("unchecked")
     public void testEmpty() {
-        Observable<Boolean> obs = Observable.<String>empty().all(hasLength3);
+        Observable<String> obs = Observable.empty();
 
         Observer<Boolean> observer = mock(Observer.class);
-        obs.subscribe(new TestSubscriber<Boolean>(observer));
+        obs.all(new Func1<String, Boolean>() {
+            @Override
+            public Boolean call(String s) {
+                return s.length() == 3;
+            }
+        }).subscribe(observer);
 
         verify(observer).onNext(true);
         verify(observer).onCompleted();
@@ -80,10 +85,15 @@ public void testEmpty() {
     @SuppressWarnings("unchecked")
     public void testError() {
         Throwable error = new Throwable();
-        Observable<Boolean> obs = Observable.<String>error(error).all(hasLength3);
+        Observable<String> obs = Observable.error(error);
 
         Observer<Boolean> observer = mock(Observer.class);
-        obs.subscribe(new TestSubscriber<Boolean>(observer));
+        obs.all(new Func1<String, Boolean>() {
+            @Override
+            public Boolean call(String s) {
+                return s.length() == 3;
+            }
+        }).subscribe(observer);
 
         verify(observer).onError(error);
         verifyNoMoreInteractions(observer);