OperationTakeWhile { public static <T> OnSubscribeFunc<T> takeWhile(final Observable<? extends T> items, final Func1<? super T, Boolean> predicate) { return takeWhileWithIndex(items, OperationTakeWhile.<T> skipIndex(predicate)); }  static OnSubscribeFunc<T> takeWhile(final Observable<? extends T> items, final Func1<? super T, Boolean> predicate); static OnSubscribeFunc<T> takeWhileWithIndex(final Observable<? extends T> items, final Func2<? super T, ? super Integer, Boolean> predicate);  }
@Test public void testTakeWhile1() { Observable<Integer> w = Observable.from(1, 2, 3); Observable<Integer> take = Observable.create(takeWhile(w, new Func1<Integer, Boolean>() { @Override public Boolean call(Integer input) { return input < 3; } })); @SuppressWarnings("unchecked") Observer<Integer> observer = mock(Observer.class); take.subscribe(observer); verify(observer, times(1)).onNext(1); verify(observer, times(1)).onNext(2); verify(observer, never()).onNext(3); verify(observer, never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); }
@Test public void testTakeWhileOnSubject1() { Subject<Integer, Integer> s = PublishSubject.create(); Observable<Integer> take = Observable.create(takeWhile(s, new Func1<Integer, Boolean>() { @Override public Boolean call(Integer input) { return input < 3; } })); @SuppressWarnings("unchecked") Observer<Integer> observer = mock(Observer.class); take.subscribe(observer); s.onNext(1); s.onNext(2); s.onNext(3); s.onNext(4); s.onNext(5); s.onCompleted(); verify(observer, times(1)).onNext(1); verify(observer, times(1)).onNext(2); verify(observer, never()).onNext(3); verify(observer, never()).onNext(4); verify(observer, never()).onNext(5); verify(observer, never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); }
@Test public void testTakeWhileDoesntLeakErrors() { Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { observer.onNext("one"); observer.onError(new Throwable("test failed")); return Subscriptions.empty(); } }); Observable.create(takeWhile(source, new Func1<String, Boolean>() { @Override public Boolean call(String s) { return false; } })).toBlockingObservable().lastOrDefault(""); }
@Test public void testTakeWhileProtectsPredicateCall() { TestObservable source = new TestObservable(mock(Subscription.class), "one"); final RuntimeException testException = new RuntimeException("test exception"); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); Observable<String> take = Observable.create(takeWhile(Observable.create(source), new Func1<String, Boolean>() { @Override public Boolean call(String s) { throw testException; } })); take.subscribe(observer); try { source.t.join(); } catch (Throwable e) { e.printStackTrace(); fail(e.getMessage()); } verify(observer, never()).onNext(any(String.class)); verify(observer, times(1)).onError(testException); }