RxJava 条件操作符和布尔操作符

一、条件操作符

• amb() 给定多个 Observable,只让第一个发射数据的 Observable 发射全部数据。

    Observable.ambArray(
                Observable.just(1,2,3,4,5).delay(1000, TimeUnit.SECONDS),
                Observable.just(6,7)
        ).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                System.out.println(integer);
            }
        });
执行结果
6
7

• defaultEmpty() 发射来自原始 Observable 的数据，如果 原始 Observable 没有发射数据，则发射一个默认数据。

   Observable.empty()
                .defaultIfEmpty(8)
                .subscribe(new Consumer<Object>() {
                    @Override
                    public void accept(Object o) throws Exception {
                        System.out.println(o);
                    }
                });
执行结果
8

• skipUntil() 丢弃原始 Observable 发射的数据，直到第二个 Observable 发射一个数据，然后发射原始 Observable 的剩余的数据。

Observable.intervalRange(1,9,0,1,TimeUnit.MILLISECONDS)
                    .skipUntil(Observable.timer(4,TimeUnit.MILLISECONDS))
                    .subscribe(new Consumer<Long>() {
                        @Override
                        public void accept(Long aLong) throws Exception {
                            System.out.println(String.valueOf(aLong));
                        }
                    });
  try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
执行结果

6
7
8
9

• skipWhile() 丢弃原始的 Observable 发射的数据，直到一个特定的条件为假，然后发射原始 Observable 剩余数据。

Observable.just(1,2,3,4,5)
                 .skipWhile(new Predicate<Integer>() {
                     @Override
                     public boolean test(Integer integer) throws Exception {
                         return integer <= 2;
                     }
                 })
                 .subscribe(new Consumer<Integer>() {
             @Override
             public void accept(Integer integer) throws Exception {
                 System.out.println(integer);
             }
         });

• takeUntil() 发射来自原始 Observable 的数据，直到第二个 Observable 发射了一个数据或一个通知。

• takeWhile() and takeWhileWithIndex() 发射原始数据 Observable 的数据，直到一个特定的条件为真，然后跳过剩余的数据。

二、布尔操作符

all() 判断是否所有的数据项都满足某个条件。

   Observable.just(1,2,3,4,5)
                .all(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer < 10;
                    }
                }).subscribe(new Consumer<Boolean>() {
            @Override
            public void accept(Boolean aBoolean) throws Exception {
                System.out.println(aBoolean);
            }
        });

执行结果
true

contains() 判断 Observable 是否会发射一个指定的值。

 Observable.just(1,2,3,4,5)
                .contains(3)
                .subscribe(new Consumer<Boolean>() {
            @Override
            public void accept(Boolean aBoolean) throws Exception {
                System.out.println(aBoolean);
            }
        });
执行结果
true

exists() and isEmpty() 判断 Observable 是否发射了一个值。

sequenceEqual() 判断两个 Observable 发射的序列是否相等。

        Observable.sequenceEqual(
                Observable.just(1,2),
                Observable.just(1,2)
        ).subscribe(new Consumer<Boolean>() {
            @Override
            public void accept(Boolean aBoolean) throws Exception {
                System.out.println(aBoolean);
            }
        });

执行结果
true