OperatorTakeWhile

Author: zsxwing

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

@@ -70,7 +70,7 @@
 import rx.operators.OperationTakeLast;
 import rx.operators.OperationTakeTimed;
 import rx.operators.OperationTakeUntil;
-import rx.operators.OperationTakeWhile;
+import rx.operators.OperatorTakeWhile;
 import rx.operators.OperationThrottleFirst;
 import rx.operators.OperationTimeInterval;
 import rx.operators.OperationTimer;
@@ -6695,7 +6695,7 @@ public final <E> Observable<T> takeUntil(Observable<? extends E> other) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Conditional-and-Boolean-Operators#wiki-takewhile-and-takewhilewithindex">RxJava Wiki: takeWhile()</a>
      */
     public final Observable<T> takeWhile(final Func1<? super T, Boolean> predicate) {
-        return create(OperationTakeWhile.takeWhile(this, predicate));
+        return lift(new OperatorTakeWhile(predicate));
     }
 
     /**
@@ -6714,7 +6714,7 @@ public final Observable<T> takeWhile(final Func1<? super T, Boolean> predicate)
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Conditional-and-Boolean-Operators#wiki-takewhile-and-takewhilewithindex">RxJava Wiki: takeWhileWithIndex()</a>
      */
     public final Observable<T> takeWhileWithIndex(final Func2<? super T, ? super Integer, Boolean> predicate) {
-        return create(OperationTakeWhile.takeWhileWithIndex(this, predicate));
+        return lift(new OperatorTakeWhile(predicate));
     }
 
     /**

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

@@ -0,0 +1,81 @@
+/**
+ * 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 rx.Observable.Operator;
+import rx.Subscriber;
+import rx.functions.Func1;
+import rx.functions.Func2;
+
+/**
+ * Returns an Observable that emits items emitted by the source Observable as long as a specified
+ * condition is true.
+ * <p>
+ * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/takeWhile.png">
+ */
+public final class OperatorTakeWhile<T> implements Operator<T, T> {
+
+    private final Func2<? super T, ? super Integer, Boolean> predicate;
+
+    public OperatorTakeWhile(final Func1<? super T, Boolean> underlying) {
+        this(new Func2<T, Integer, Boolean>() {
+            @Override
+            public Boolean call(T input, Integer index) {
+                return underlying.call(input);
+            }
+        });
+    }
+
+    public OperatorTakeWhile(Func2<? super T, ? super Integer, Boolean> predicate) {
+        this.predicate = predicate;
+    }
+
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super T> subscriber) {
+        return new Subscriber<T>(subscriber) {
+
+            private int counter = 0;
+
+            @Override
+            public void onNext(T args) {
+                boolean isSelected;
+                try {
+                    isSelected = predicate.call(args, counter++);
+                } catch (Throwable e) {
+                    subscriber.onError(e);
+                    return;
+                }
+                if (isSelected) {
+                    subscriber.onNext(args);
+                } else {
+                    subscriber.onCompleted();
+                }
+            }
+
+            @Override
+            public void onCompleted() {
+                subscriber.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                subscriber.onError(e);
+            }
+
+        };
+    }
+
+}

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

@@ -21,31 +21,30 @@
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
-import static rx.operators.OperationTakeWhile.takeWhile;
-import static rx.operators.OperationTakeWhile.takeWhileWithIndex;
 
 import org.junit.Test;
 
 import rx.Observable;
+import rx.Observable.OnSubscribe;
 import rx.Observer;
+import rx.Subscriber;
 import rx.Subscription;
 import rx.functions.Func1;
 import rx.functions.Func2;
 import rx.subjects.PublishSubject;
 import rx.subjects.Subject;
-import rx.subscriptions.Subscriptions;
 
-public class OperationTakeWhileTest {
+public class OperatorTakeWhileTest {
 
     @Test
     public void testTakeWhile1() {
         Observable<Integer> w = Observable.from(1, 2, 3);
-        Observable<Integer> take = Observable.create(takeWhile(w, new Func1<Integer, Boolean>() {
+        Observable<Integer> take = w.takeWhile(new Func1<Integer, Boolean>() {
             @Override
             public Boolean call(Integer input) {
                 return input < 3;
             }
-        }));
+        });
 
         @SuppressWarnings("unchecked")
         Observer<Integer> observer = mock(Observer.class);
@@ -60,12 +59,12 @@ public Boolean call(Integer input) {
     @Test
     public void testTakeWhileOnSubject1() {
         Subject<Integer, Integer> s = PublishSubject.create();
-        Observable<Integer> take = Observable.create(takeWhile(s, new Func1<Integer, Boolean>() {
+        Observable<Integer> take = s.takeWhile(new Func1<Integer, Boolean>() {
             @Override
             public Boolean call(Integer input) {
                 return input < 3;
             }
-        }));
+        });
 
         @SuppressWarnings("unchecked")
         Observer<Integer> observer = mock(Observer.class);
@@ -90,12 +89,12 @@ public Boolean call(Integer input) {
     @Test
     public void testTakeWhile2() {
         Observable<String> w = Observable.from("one", "two", "three");
-        Observable<String> take = Observable.create(takeWhileWithIndex(w, new Func2<String, Integer, Boolean>() {
+        Observable<String> take = w.takeWhileWithIndex(new Func2<String, Integer, Boolean>() {
             @Override
             public Boolean call(String input, Integer index) {
                 return index < 2;
             }
-        }));
+        });
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
@@ -109,21 +108,20 @@ public Boolean call(String input, Integer index) {
 
     @Test
     public void testTakeWhileDoesntLeakErrors() {
-        Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() {
+        Observable<String> source = Observable.create(new OnSubscribe<String>() {
             @Override
-            public Subscription onSubscribe(Observer<? super String> observer) {
+            public void call(Subscriber<? super String> observer) {
                 observer.onNext("one");
                 observer.onError(new Throwable("test failed"));
-                return Subscriptions.empty();
             }
         });
 
-        Observable.create(takeWhile(source, new Func1<String, Boolean>() {
+        source.takeWhile(new Func1<String, Boolean>() {
             @Override
             public Boolean call(String s) {
                 return false;
             }
-        })).toBlockingObservable().lastOrDefault("");
+        }).toBlockingObservable().lastOrDefault("");
     }
 
     @Test
@@ -133,12 +131,12 @@ public void testTakeWhileProtectsPredicateCall() {
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
-        Observable<String> take = Observable.create(takeWhile(Observable.create(source), new Func1<String, Boolean>() {
+        Observable<String> take = Observable.create(source).takeWhile(new Func1<String, Boolean>() {
             @Override
             public Boolean call(String s) {
                 throw testException;
             }
-        }));
+        });
         take.subscribe(observer);
 
         // wait for the Observable to complete
@@ -160,12 +158,12 @@ public void testUnsubscribeAfterTake() {
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
-        Observable<String> take = Observable.create(takeWhileWithIndex(Observable.create(w), new Func2<String, Integer, Boolean>() {
+        Observable<String> take = Observable.create(w).takeWhileWithIndex(new Func2<String, Integer, Boolean>() {
             @Override
             public Boolean call(String s, Integer index) {
                 return index < 1;
             }
-        }));
+        });
         take.subscribe(observer);
 
         // wait for the Observable to complete