RxJava操作符实践：7_条件和布尔操作之5_takeUntil

一、描述

当第二个Observable发射了一项数据或者终止时，丢弃原始Observable发射的任何数据。

TakeUntil订阅并开始发射原始Observable，它还监视你提供的第二个Observable。如果第二个Observable发射了一项数据或者发射了一个终止通知，TakeUntil返回的Observable会停止发射原始Observable并终止。

RxJava中的实现是takeUntil。注意：第二个Observable发射一项数据或一个onError通知或一个onCompleted通知都会导致takeUntil停止发射数据。

takeUntil默认不在任何特定的调度器上执行。

二、示意图

三、示例代码

Observable observable1 = Observable.interval(1000, TimeUnit.MILLISECONDS).take(5);
Observable observable2 = Observable.just(1, 2).delay(2500, TimeUnit.MILLISECONDS);

observable1.takeUntil(observable2)
    .subscribe(new Subscriber<Long>() {
        @Override
        public void onCompleted() {
            System.out.println("onCompleted.");
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("onError: " + e.getMessage());
        }

        @Override
        public void onNext(Long aLong) {
            System.out.println("onNext: " + aLong);
        }
    });

四、运行结果

onNext: 0
onNext: 1
onCompleted.

1. observable1每隔1秒发一个数据（第1秒时发射0，第2秒时发射1），takeUntil正常把数据发射出来；
2. 直到observable2在2.5秒的时候发射了第1个数据；
3. takeUntil就停止发射数据了。

五、参考资料

ReactiveX官方文档

ReactiveX文档中文翻译

PS：欢迎关注 SherlockShi 个人博客