OperationTakeUntil { public static <T, E> Observable<T> takeUntil(final Observable<? extends T> source, final Observable<? extends E> other) { Observable<Notification<T>> s = Observable.create(new SourceObservable<T>(source)); Observable<Notification<T>> o = Observable.create(new OtherObservable<T, E>(other)); Observable<Notification<T>> result = Observable.merge(s, o); return result.takeWhile(new Func1<Notification<T>, Boolean>() { @Override public Boolean call(Notification<T> notification) { return !notification.halt; } }).map(new Func1<Notification<T>, T>() { @Override public T call(Notification<T> notification) { return notification.value; } }); }  static Observable<T> takeUntil(final Observable<? extends T> source, final Observable<? extends E> other);  }
@Test @SuppressWarnings("unchecked") public void testTakeUntil() { Subscription sSource = mock(Subscription.class); Subscription sOther = mock(Subscription.class); TestObservable source = new TestObservable(sSource); TestObservable other = new TestObservable(sOther); Observer<String> result = mock(Observer.class); Observable<String> stringObservable = takeUntil(Observable.create(source), Observable.create(other)); stringObservable.subscribe(result); source.sendOnNext("one"); source.sendOnNext("two"); other.sendOnNext("three"); source.sendOnNext("four"); source.sendOnCompleted(); other.sendOnCompleted(); verify(result, times(1)).onNext("one"); verify(result, times(1)).onNext("two"); verify(result, times(0)).onNext("three"); verify(result, times(0)).onNext("four"); verify(sSource, times(1)).unsubscribe(); verify(sOther, times(1)).unsubscribe(); }
@Test @SuppressWarnings("unchecked") public void testTakeUntilSourceCompleted() { Subscription sSource = mock(Subscription.class); Subscription sOther = mock(Subscription.class); TestObservable source = new TestObservable(sSource); TestObservable other = new TestObservable(sOther); Observer<String> result = mock(Observer.class); Observable<String> stringObservable = takeUntil(Observable.create(source), Observable.create(other)); stringObservable.subscribe(result); source.sendOnNext("one"); source.sendOnNext("two"); source.sendOnCompleted(); verify(result, times(1)).onNext("one"); verify(result, times(1)).onNext("two"); verify(sSource, times(1)).unsubscribe(); verify(sOther, times(1)).unsubscribe(); }
@Test @SuppressWarnings("unchecked") public void testTakeUntilSourceError() { Subscription sSource = mock(Subscription.class); Subscription sOther = mock(Subscription.class); TestObservable source = new TestObservable(sSource); TestObservable other = new TestObservable(sOther); Throwable error = new Throwable(); Observer<String> result = mock(Observer.class); Observable<String> stringObservable = takeUntil(Observable.create(source), Observable.create(other)); stringObservable.subscribe(result); source.sendOnNext("one"); source.sendOnNext("two"); source.sendOnError(error); source.sendOnNext("three"); verify(result, times(1)).onNext("one"); verify(result, times(1)).onNext("two"); verify(result, times(0)).onNext("three"); verify(result, times(1)).onError(error); verify(sSource, times(1)).unsubscribe(); verify(sOther, times(1)).unsubscribe(); }
@Test @SuppressWarnings("unchecked") public void testTakeUntilOtherError() { Subscription sSource = mock(Subscription.class); Subscription sOther = mock(Subscription.class); TestObservable source = new TestObservable(sSource); TestObservable other = new TestObservable(sOther); Throwable error = new Throwable(); Observer<String> result = mock(Observer.class); Observable<String> stringObservable = takeUntil(Observable.create(source), Observable.create(other)); stringObservable.subscribe(result); source.sendOnNext("one"); source.sendOnNext("two"); other.sendOnError(error); source.sendOnNext("three"); verify(result, times(1)).onNext("one"); verify(result, times(1)).onNext("two"); verify(result, times(0)).onNext("three"); verify(result, times(1)).onError(error); verify(result, times(0)).onCompleted(); verify(sSource, times(1)).unsubscribe(); verify(sOther, times(1)).unsubscribe(); }
@Test @SuppressWarnings("unchecked") public void testTakeUntilOtherCompleted() { Subscription sSource = mock(Subscription.class); Subscription sOther = mock(Subscription.class); TestObservable source = new TestObservable(sSource); TestObservable other = new TestObservable(sOther); Observer<String> result = mock(Observer.class); Observable<String> stringObservable = takeUntil(Observable.create(source), Observable.create(other)); stringObservable.subscribe(result); source.sendOnNext("one"); source.sendOnNext("two"); other.sendOnCompleted(); source.sendOnNext("three"); verify(result, times(1)).onNext("one"); verify(result, times(1)).onNext("two"); verify(result, times(0)).onNext("three"); verify(result, times(1)).onCompleted(); verify(sSource, times(1)).unsubscribe(); verify(sOther, times(1)).unsubscribe(); }