@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(); }
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; } }); }
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; } }); } }
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; } }); }  }
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); }
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);  }