Changed operator names, fixed backpressure.

Author: akarnokd

src/main/java/rx/Observable.java

@@ -7840,8 +7840,8 @@ public final Observable<T> takeWhile(final Func1<? super T, Boolean> predicate)
      * @see Observable#takeWhile(Func1)
      */
     @Experimental
-    public final Observable<T> doTakeWhile(final Func1<? super T, Boolean> predicate) {
-        return lift(new OperatorDoTakeWhile<T>(predicate));
+    public final Observable<T> takeUntil(final Func1<? super T, Boolean> predicate) {
+        return lift(new OperatorTakeUntilPredicate<T>(predicate));
     }
 
     /**

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

@@ -0,0 +1,97 @@
+/**
+ * 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.internal.operators;
+
+import rx.Observable.Operator;
+import rx.*;
+import rx.annotations.Experimental;
+import rx.functions.Func1;
+
+/**
+ * Returns an Observable that emits items emitted by the source Observable until
+ * the provided predicate returns false
+ * <p>
+ */
+@Experimental
+public final class OperatorTakeUntilPredicate<T> implements Operator<T, T> {
+    /** Subscriber returned to the upstream. */
+    private final class ParentSubscriber extends Subscriber<T> {
+        private final Subscriber<? super T> child;
+        private boolean done = false;
+
+        private ParentSubscriber(Subscriber<? super T> child) {
+            this.child = child;
+        }
+
+        @Override
+        public void onNext(T args) {
+            child.onNext(args);
+            
+            boolean doContinue = false;
+            try {
+                doContinue = predicate.call(args);
+            } catch (Throwable e) {
+                done = true;
+                child.onError(e);
+                unsubscribe();
+                return;
+            }
+            if (!doContinue) {
+                done = true;
+                child.onCompleted();
+                unsubscribe();
+            }
+        }
+
+        @Override
+        public void onCompleted() {
+            if (!done) {
+                child.onCompleted();
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            if (!done) {
+                child.onError(e);
+            }
+        }
+        void downstreamRequest(long n) {
+            request(n);
+        }
+    }
+
+    private final Func1<? super T, Boolean> predicate;
+
+    public OperatorTakeUntilPredicate(final Func1<? super T, Boolean> predicate) {
+        this.predicate = predicate;
+    }
+
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super T> child) {
+        final ParentSubscriber parent = new ParentSubscriber(child);
+        child.add(parent); // don't unsubscribe downstream
+        child.setProducer(new Producer() {
+            @Override
+            public void request(long n) {
+                parent.downstreamRequest(n);
+            }
+        });
+        
+        return parent;
+    }
+
+}

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

@@ -19,21 +19,24 @@
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.*;
 
-import org.junit.Test;
+import java.util.Arrays;
+
+import org.junit.*;
 
 import rx.*;
 import rx.exceptions.TestException;
 import rx.functions.Func1;
 import rx.internal.util.UtilityFunctions;
+import rx.observers.TestSubscriber;
 ;
 
-public class OperatorDoTakeWhileTest {
+public class OperatorTakeUntilPredicateTest {
     @Test
     public void takeEmpty() {
         @SuppressWarnings("unchecked")
         Observer<Object> o = mock(Observer.class);
 
-        Observable.empty().doTakeWhile(UtilityFunctions.alwaysTrue()).subscribe(o);
+        Observable.empty().takeUntil(UtilityFunctions.alwaysTrue()).subscribe(o);
 
         verify(o, never()).onNext(any());
         verify(o, never()).onError(any(Throwable.class));
@@ -44,7 +47,7 @@ public void takeAll() {
         @SuppressWarnings("unchecked")
         Observer<Object> o = mock(Observer.class);
 
-        Observable.just(1, 2).doTakeWhile(UtilityFunctions.alwaysTrue()).subscribe(o);
+        Observable.just(1, 2).takeUntil(UtilityFunctions.alwaysTrue()).subscribe(o);
 
         verify(o).onNext(1);
         verify(o).onNext(2);
@@ -56,7 +59,7 @@ public void takeFirst() {
         @SuppressWarnings("unchecked")
         Observer<Object> o = mock(Observer.class);
 
-        Observable.just(1, 2).doTakeWhile(UtilityFunctions.alwaysFalse()).subscribe(o);
+        Observable.just(1, 2).takeUntil(UtilityFunctions.alwaysFalse()).subscribe(o);
 
         verify(o).onNext(1);
         verify(o, never()).onNext(2);
@@ -68,7 +71,7 @@ public void takeSome() {
         @SuppressWarnings("unchecked")
         Observer<Object> o = mock(Observer.class);
 
-        Observable.just(1, 2, 3).doTakeWhile(new Func1<Integer, Boolean>() {
+        Observable.just(1, 2, 3).takeUntil(new Func1<Integer, Boolean>() {
             @Override
             public Boolean call(Integer t1) {
                 return t1 < 2;
@@ -86,7 +89,7 @@ public void functionThrows() {
         @SuppressWarnings("unchecked")
         Observer<Object> o = mock(Observer.class);
 
-        Observable.just(1, 2, 3).doTakeWhile(new Func1<Integer, Boolean>() {
+        Observable.just(1, 2, 3).takeUntil(new Func1<Integer, Boolean>() {
             @Override
             public Boolean call(Integer t1) {
                 throw new TestException("Forced failure");
@@ -107,11 +110,26 @@ public void sourceThrows() {
         Observable.just(1)
         .concatWith(Observable.<Integer>error(new TestException()))
         .concatWith(Observable.just(2))
-        .doTakeWhile(UtilityFunctions.alwaysTrue()).subscribe(o);
+        .takeUntil(UtilityFunctions.alwaysTrue()).subscribe(o);
 
         verify(o).onNext(1);
         verify(o, never()).onNext(2);
         verify(o).onError(any(TestException.class));
         verify(o, never()).onCompleted();
     }
+    @Test
+    public void backpressure() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
+            @Override
+            public void onStart() {
+                request(5);
+            }
+        };
+        
+        Observable.range(1, 1000).takeUntil(UtilityFunctions.alwaysTrue()).subscribe(ts);
+        
+        ts.assertNoErrors();
+        ts.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5));
+        Assert.assertEquals(0, ts.getOnCompletedEvents().size());
+    }
 }