Changed predicate to stopPredicate: stops on becoming true.

Author: akarnokd

src/main/java/rx/Observable.java

@@ -7827,21 +7827,21 @@ public final Observable<T> takeWhile(final Func1<? super T, Boolean> predicate)
     /**
      * Returns an Observable that first emits items emitted by the source Observable, 
      * checks the specified condition after each item, and
-     * then completes as soon as this condition is not satisfied.
+     * then completes if the condition is satisfied.
      * <p>
      * The difference between this operator and {@link #takeWhile(Func1)} is that here, the condition is evaluated <b>after</b>
      * the item was emitted.
      * 
-     * @param predicate 
+     * @param stopPredicate 
      *            a function that evaluates an item emitted by the source Observable and returns a Boolean
      * @return an Observable that first emits items emitted by the source Observable, 
      *         checks the specified condition after each item, and
-     *         then completes as soon as this condition is not satisfied.
+     *         then completes if the condition is satisfied.
      * @see Observable#takeWhile(Func1)
      */
     @Experimental
-    public final Observable<T> takeUntil(final Func1<? super T, Boolean> predicate) {
-        return lift(new OperatorTakeUntilPredicate<T>(predicate));
+    public final Observable<T> takeUntil(final Func1<? super T, Boolean> stopPredicate) {
+        return lift(new OperatorTakeUntilPredicate<T>(stopPredicate));
     }
 
     /**

src/main/java/rx/internal/operators/OperatorTakeUntilPredicate.java

@@ -40,16 +40,16 @@ private ParentSubscriber(Subscriber<? super T> child) {
         public void onNext(T args) {
             child.onNext(args);
 
-            boolean doContinue = false;
+            boolean stop = false;
             try {
-                doContinue = predicate.call(args);
+                stop = stopPredicate.call(args);
             } catch (Throwable e) {
                 done = true;
                 child.onError(e);
                 unsubscribe();
                 return;
             }
-            if (!doContinue) {
+            if (stop) {
                 done = true;
                 child.onCompleted();
                 unsubscribe();
@@ -74,10 +74,10 @@ void downstreamRequest(long n) {
         }
     }
 
-    private final Func1<? super T, Boolean> predicate;
+    private final Func1<? super T, Boolean> stopPredicate;
 
-    public OperatorTakeUntilPredicate(final Func1<? super T, Boolean> predicate) {
-        this.predicate = predicate;
+    public OperatorTakeUntilPredicate(final Func1<? super T, Boolean> stopPredicate) {
+        this.stopPredicate = stopPredicate;
     }
 
     @Override

src/test/java/rx/internal/operators/OperatorTakeUntilPredicateTest.java

@@ -47,7 +47,7 @@ public void takeAll() {
         @SuppressWarnings("unchecked")
         Observer<Object> o = mock(Observer.class);
 
-        Observable.just(1, 2).takeUntil(UtilityFunctions.alwaysTrue()).subscribe(o);
+        Observable.just(1, 2).takeUntil(UtilityFunctions.alwaysFalse()).subscribe(o);
 
         verify(o).onNext(1);
         verify(o).onNext(2);
@@ -59,7 +59,7 @@ public void takeFirst() {
         @SuppressWarnings("unchecked")
         Observer<Object> o = mock(Observer.class);
 
-        Observable.just(1, 2).takeUntil(UtilityFunctions.alwaysFalse()).subscribe(o);
+        Observable.just(1, 2).takeUntil(UtilityFunctions.alwaysTrue()).subscribe(o);
 
         verify(o).onNext(1);
         verify(o, never()).onNext(2);
@@ -74,7 +74,7 @@ public void takeSome() {
         Observable.just(1, 2, 3).takeUntil(new Func1<Integer, Boolean>() {
             @Override
             public Boolean call(Integer t1) {
-                return t1 < 2;
+                return t1 == 2;
             }
         }).subscribe(o);
 
@@ -110,7 +110,7 @@ public void sourceThrows() {
         Observable.just(1)
         .concatWith(Observable.<Integer>error(new TestException()))
         .concatWith(Observable.just(2))
-        .takeUntil(UtilityFunctions.alwaysTrue()).subscribe(o);
+        .takeUntil(UtilityFunctions.alwaysFalse()).subscribe(o);
 
         verify(o).onNext(1);
         verify(o, never()).onNext(2);
@@ -126,7 +126,7 @@ public void onStart() {
             }
         };
 
-        Observable.range(1, 1000).takeUntil(UtilityFunctions.alwaysTrue()).subscribe(ts);
+        Observable.range(1, 1000).takeUntil(UtilityFunctions.alwaysFalse()).subscribe(ts);
 
         ts.assertNoErrors();
         ts.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5));