Merge pull request #2493 from akarnokd/OperatorDoTakeWhile

Operator TakeUntil with predicate

Author: benjchristensen

src/main/java/rx/Observable.java

@@ -7854,6 +7854,26 @@ public final Observable<T> takeWhile(final Func1<? super T, Boolean> predicate)
         return lift(new OperatorTakeWhile<T>(predicate));
     }
 
+    /**
+     * Returns an Observable that first emits items emitted by the source Observable, 
+     * checks the specified condition after each item, and
+     * 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 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 if the condition is satisfied.
+     * @see Observable#takeWhile(Func1)
+     */
+    @Experimental
+    public final Observable<T> takeUntil(final Func1<? super T, Boolean> stopPredicate) {
+        return lift(new OperatorTakeUntilPredicate<T>(stopPredicate));
+    }
+    
     /**
      * Returns an Observable that emits only the first item emitted by the source Observable during sequential
      * time windows of a specified duration.

src/main/java/rx/internal/operators/OperatorTakeUntilPredicate.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 stop = false;
+            try {
+                stop = stopPredicate.call(args);
+            } catch (Throwable e) {
+                done = true;
+                child.onError(e);
+                unsubscribe();
+                return;
+            }
+            if (stop) {
+                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> stopPredicate;
+
+    public OperatorTakeUntilPredicate(final Func1<? super T, Boolean> stopPredicate) {
+        this.stopPredicate = stopPredicate;
+    }
+
+    @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/test/java/rx/internal/operators/OperatorTakeUntilPredicateTest.java

@@ -0,0 +1,135 @@
+/**
+ * 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 static org.mockito.Matchers.any;
+import static org.mockito.Mockito.*;
+
+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 OperatorTakeUntilPredicateTest {
+    @Test
+    public void takeEmpty() {
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        
+        Observable.empty().takeUntil(UtilityFunctions.alwaysTrue()).subscribe(o);
+        
+        verify(o, never()).onNext(any());
+        verify(o, never()).onError(any(Throwable.class));
+        verify(o).onCompleted();
+    }
+    @Test
+    public void takeAll() {
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        
+        Observable.just(1, 2).takeUntil(UtilityFunctions.alwaysFalse()).subscribe(o);
+        
+        verify(o).onNext(1);
+        verify(o).onNext(2);
+        verify(o, never()).onError(any(Throwable.class));
+        verify(o).onCompleted();
+    }
+    @Test
+    public void takeFirst() {
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        
+        Observable.just(1, 2).takeUntil(UtilityFunctions.alwaysTrue()).subscribe(o);
+        
+        verify(o).onNext(1);
+        verify(o, never()).onNext(2);
+        verify(o, never()).onError(any(Throwable.class));
+        verify(o).onCompleted();
+    }
+    @Test
+    public void takeSome() {
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        
+        Observable.just(1, 2, 3).takeUntil(new Func1<Integer, Boolean>() {
+            @Override
+            public Boolean call(Integer t1) {
+                return t1 == 2;
+            }
+        }).subscribe(o);
+        
+        verify(o).onNext(1);
+        verify(o).onNext(2);
+        verify(o, never()).onNext(3);
+        verify(o, never()).onError(any(Throwable.class));
+        verify(o).onCompleted();
+    }
+    @Test
+    public void functionThrows() {
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        
+        Observable.just(1, 2, 3).takeUntil(new Func1<Integer, Boolean>() {
+            @Override
+            public Boolean call(Integer t1) {
+                throw new TestException("Forced failure");
+            }
+        }).subscribe(o);
+        
+        verify(o).onNext(1);
+        verify(o, never()).onNext(2);
+        verify(o, never()).onNext(3);
+        verify(o).onError(any(TestException.class));
+        verify(o, never()).onCompleted();
+    }
+    @Test
+    public void sourceThrows() {
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        
+        Observable.just(1)
+        .concatWith(Observable.<Integer>error(new TestException()))
+        .concatWith(Observable.just(2))
+        .takeUntil(UtilityFunctions.alwaysFalse()).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.alwaysFalse()).subscribe(ts);
+        
+        ts.assertNoErrors();
+        ts.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5));
+        Assert.assertEquals(0, ts.getOnCompletedEvents().size());
+    }
+}