RxJava操作符——条件和布尔操作符（Conditional and Boolean Operators）

RxJava系列教程：

1. RxJava使用介绍 【视频教程】
2. RxJava操作符
  ? Creating Observables(Observable的创建操作符) 【视频教程】
  ? Transforming Observables(Observable的转换操作符) 【视频教程】
  ? Filtering Observables(Observable的过滤操作符) 【视频教程】
  ? Combining Observables(Observable的组合操作符) 【视频教程】
  ? Error Handling Operators(Observable的错误处理操作符) 【视频教程】
  ? Observable Utility Operators(Observable的辅助性操作符) 【视频教程】
  ? Conditional and Boolean Operators(Observable的条件和布尔操作符) 【视频教程】
  ? Mathematical and Aggregate Operators(Observable数学运算及聚合操作符) 【视频教程】
  ? 其他如observable.toList()、observable.connect()、observable.publish()等等； 【视频教程】
3. RxJava Observer与Subcriber的关系 【视频教程】
4. RxJava线程控制（Scheduler） 【视频教程】
5. RxJava 并发之数据流发射太快如何办（背压（Backpressure）） 【视频教程】

All

All操作符根据一个函数对源Observable发射的所有数据进行判断，最终返回的结果就是这个判断结果。这个函数使用发射的数据作为参数，内部判断所有的数据是否满足我们定义好的判断条件，如果全部都满足则返回true，否则就返回false。

示例代码：

Observable.from(studentList).all(new Func1<Student, Boolean>() {

            @Override
            public Boolean call(Student student) {
                return student.age > 25;
            }
        }).subscribe(new Action1<Boolean>() {

            @Override
            public void call(Boolean value) {
                System.out.println("onSuccess value = " + value);
            }
        });

List<Student> studentList = new ArrayList<Student>() {
        {
            add(new Student("Stay", 28));
            add(new Student("谷歌小弟", 23));
            add(new Student("Star", 25));
        }
    };

输出：

onSuccess value = false

Amb

传递两个或多个（至多9个）Observable给Amb时，它只发射其中首先发射数据或通知（onError或onCompleted）的那个Observable的所有数据，而其他所有的Observable的发射物将被丢弃。

有一个类似的对象方法ambWith。Observable.amb(o1,o2)和o1.ambWith(o2)是等价的。

示例代码：

Observable.amb(Observable.just(1,2,3,4,5).delay(2, TimeUnit.SECONDS),
                Observable.just(100,101))
                .subscribe(new Action1<Integer>() {

                    @Override
                    public void call(Integer value) {
                        System.out.println("call value = " + value);
                    }
                });

输出：

call value = 100
call value = 101

Contains、IsEmpty 、exists

Contains操作符用来判断源Observable所发射的数据是否包含某一个数据，如果包含会返回true，如果源Observable已经结束了却还没有发射这个数据则返回false。

IsEmpty：相关的一个操作符IsEmpty用于判定原始Observable是否没有发射任何数据。

exists：RxJava中还有一个exists操作符，它通过一个谓词函数测试原始Observable发射的数据，只要任何一项满足条件就返回一个发射true的Observable，否则返回一个发射false的Observable。

示例代码：

//Contains:判定一个Observable是否发射一个特定的值
Observable.just(4,5,6)
        .contains(4)
        .subscribe(new Action1<Boolean>() {
            @Override
            public void call(Boolean aBoolean) {
                System.out.println("contains(4):"+aBoolean);
            }
        });
//isEmpty:判定原始Observable是否没有发射任何数据
Observable.just(4,5,6)
        .isEmpty()
        .subscribe(new Action1<Boolean>() {
            @Override
            public void call(Boolean aBoolean) {
                System.out.println("isEmpty():"+aBoolean));
            }
        });

//exists操作符，它通过一个谓词函数测试原始Observable发射的数据，
// 只要任何一项满足条件就返回一个发射true的Observable，
// 否则返回一个发射false的Observable。
Observable.just(4,5,6)
        .exists(new Func1<Integer, Boolean>() {
            @Override
            public Boolean call(Integer integer) {
                return integer<5;
            }
        })
        .subscribe(new Action1<Boolean>() {
            @Override
            public void call(Boolean aBoolean) {
                System.out.println("exists():"+aBoolean);
            }
        });

输出：

contains(4):true 
isEmpty():false 
exists():true

DefaultIfEmpty

??DefaultIfEmpty简单的精确地发射原始Observable的值，如果原始Observable没有发射任何数据正常终止（以onCompleted的形式），DefaultIfEmpty返回的Observable就发射一个你提供的默认值。

示例代码：

Observable.empty()
        .defaultIfEmpty(10)
        .subscribe(new Action1<Integer>() {

            @Override
            public void call(Integer value) {
                System.out.println("defaultIfEmpty:" + value);              
            }
        });

输出：

defaultIfEmpty:10

SequenceEqual

判定两个Observables是否发射相同的数据序列。 传递两个Observable给SequenceEqual操作符，它会比较两个Observable的发射物，如果两个序列是相同的（相同的数据，相同的顺序，相同的终止状态），它就发射true，否则发射false。

示例代码：

Observable.sequenceEqual(
        //第一个Observable延迟1秒发射数据
        Observable.just(4,5,6).delay(1,TimeUnit.SECONDS),
        Observable.just(4,5,6))
        .subscribe(new Action1<Boolean>() {
            @Override
            public void call(Boolean aBoolean) {
                System.out.println("sequenceEqual:"+aBoolean);
            }
        });
        .subscribe(aBoolean -> Log.v(TAG, "sequenceEqual:"+aBoolean));

输出：

sequenceEqual:true

SkipUntil

SkipUntil订阅原始的Observable，但是忽略它的发射物，直到第二个Observable发射了一项数据那一刻，它开始发射原始Observable。

示例代码：

Observable.interval(1, TimeUnit.SECONDS) //从0开始
        .take(6)
        .skipUntil(Observable.just(10).delay(3,TimeUnit.SECONDS))
        .subscribe(new Subscriber<Long>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Long value) {
                System.out.println("onNext value = " + value);

            }
        });

输出：

skipUntil value = 3
skipUntil value = 4
skipUntil value = 5

SkipWhile

SkipWhile订阅原始的Observable，但是忽略它的发射物，直到你指定的某个条件变为false的那一刻，它开始发射原始Observable。

示例代码：

Observable.interval(1, TimeUnit.SECONDS) //从0开始
        .take(6)
        .skipWhile(new Func1<Long, Boolean>() {
            @Override
            public Boolean call(Long aLong) {
                return aLong<3;   //舍弃原Observable发射的数据，直到发射的数据>=3，才继续发射
            }
        })
        .subscribe(new Subscriber<Long>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Long value) {
                System.out.println("skipWhile value = " + value);

            }
        });

输出：

skipWhile value = 3
skipWhile value = 4
skipWhile value = 5

TakeUntil

??takeUntil与skipUntil操作符作用相反，订阅并开始发射原始Observable，它还监视你提供的第二个Observable。如果第二个Observable发射了一项数据或者发射了一个终止通知（ onError通知或一个onCompleted通知），TakeUntil返回的Observable会停止发射原始Observable并终止。

示例代码：

Observable.interval(500, TimeUnit.MILLISECONDS) //从0开始
        .takeUntil(Observable.just(100, 101, 102).delay(3, TimeUnit.SECONDS))
        .subscribe(new Subscriber<Long>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Long value) {
                System.out.println("takeUntil value = " + value);

            }
        });

输出：

takeUntil value = 0
takeUntil value = 1
takeUntil value = 2
takeUntil value = 3
takeUntil value = 4

TakeWhile

takeWhile与skipWhile操作符作用相反，发射原始Observable，直到你指定的某个条件不成立的那一刻，它停止发射原始Observable，并终止自己的Observable。

示例代码：

Observable.interval(1, TimeUnit.SECONDS) //从0开始
        .takeWhile(new Func1<Long, Boolean>() {
            @Override
            public Boolean call(Long aLong) {
                return aLong<3;   
            }
        })
        .subscribe(new Subscriber<Long>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Long value) {
                System.out.println("takeWhile value = " + value);

            }
        });

输出：

takeWhile value = 0
takeWhile value = 1
takeWhile value = 2