RxJava2.0操作符之--条件布尔操作符

all

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

Observable observable=Observable.create(new ObservableOnSubscribe() {
    public void subscribe(@NonNull ObservableEmitter e) throws Exception {
                e.onNext(2);
  e.onNext(3);
  e.onNext(4);
  e.onNext(5);
  }
});
observable.all(new Predicate() {
    public boolean test(@NonNull Integer integer) throws Exception {
        return integer>4;
  }
})
        .toObservable().subscribe(RxUtils.getObserver());
observable.subscribe(RxUtils.getObserver());

运行结果

onSubscribe
Thread:Thread[main,5,main]
onNext:false
Thread:Thread[main,5,main]
onComplete
Thread:Thread[main,5,main]
onSubscribe
Thread:Thread[main,5,main]
onNext:2
Thread:Thread[main,5,main]
onNext:3
Thread:Thread[main,5,main]
onNext:4
Thread:Thread[main,5,main]
onNext:5
Thread:Thread[main,5,main]

amb

给定多个Observable，只让第一个发射数据的Observable发射全部数据

List> list= new ArrayList>();
list.add( Observable.just(1,2,4));
list.add( Observable.just(4,5,6).delay(3,TimeUnit.SECONDS));
Observable.amb(list).subscribe(RxUtils.getObserver());

运行结果：

onSubscribe
Thread:Thread[main,5,main]
onNext:1
Thread:Thread[main,5,main]
onNext:2
Thread:Thread[main,5,main]
onNext:4
Thread:Thread[main,5,main]
onComplete
Thread:Thread[main,5,main]

可以看到，第一个源打印出，第二个源被丢弃，符合我们的预期

Contains

判断Observable是否会发射一个指定的值

Observable observable= Observable.just(1,2,3,4);
  Single bollean= observable.contains(6);
  System.out.println(bollean.blockingGet());

运行结果

false

defaultIfEmpty

发射来自原始Observable的数据，如果原始Observable没有发射数据，就发射一个默认数据

Observable.empty().defaultIfEmpty(0).subscribe(RxUtils.getObserver());

    onSubscribe
    Thread:Thread[main,5,main]
    onNext:0
    Thread:Thread[main,5,main]
    onComplete
    Thread:Thread[main,5,main]

sequenceEqual

判断两个Observables发射的序列是否相等

Observable observable=  Observable.just(1,2,3);
Observable observable2=Observable.just(1,2,3);
Observable observable3=Observable.just(4,6,7);
System.out.println(Observable.sequenceEqual(observable,observable2).blockingGet());
System.out.println(Observable.sequenceEqual(observable,observable3).blockingGet());

true
false

skipUntil( )

丢弃原始Observable发射的数据，直到第二个Observable发射了一个数据，然后发射原始Observable的剩余数据

  Observable.interval(1, TimeUnit.SECONDS).skipUntil(Observable.timer(5,TimeUnit.SECONDS)).subscribe(RxUtils.getObserver());
  try {
    Thread.sleep(Integer.MAX_VALUE);
} catch (InterruptedException e1) {
    e1.printStackTrace();
}

    onSubscribe
    Thread:Thread[main,5,main]
    onNext:4
    Thread:Thread[RxComputationThreadPool-2,5,main]
    onNext:5
    Thread:Thread[RxComputationThreadPool-2,5,main]
    onNext:6
    Thread:Thread[RxComputationThreadPool-2,5,main]
    onNext:7
    Thread:Thread[RxComputationThreadPool-2,5,main]
    onNext:8
    Thread:Thread[RxComputationThreadPool-2,5,main]
    onNext:9
    Thread:Thread[RxComputationThreadPool-2,5,main]
    onNext:10
    Thread:Thread[RxComputationThreadPool-2,5,main]
    onNext:11
    Thread:Thread[RxComputationThreadPool-2,5,main]
    onNext:12

skipWhile

丢弃Observable发射的数据，直到一个指定的条件不成立

Observable.interval(1, TimeUnit.SECONDS).skipWhile(new Predicate() {
    public boolean test(@NonNull Long aLong) throws Exception {
        return aLong<6;
  }
}).subscribe(RxUtils.getObserver());
try {
    Thread.sleep(Integer.MAX_VALUE);
} catch (InterruptedException e1) {
    e1.printStackTrace();
}

onSubscribe
Thread:Thread[main,5,main]
onNext:6
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:7
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:8
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:9
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:10

TakeWhile

发射Observable发射的数据，直到一个指定的条件不成立

Observable.interval(1, TimeUnit.SECONDS)
        .takeWhile(new Predicate() {
            public boolean test(@NonNull Long aLong) throws Exception {
                return aLong < 5;
  }
        }).subscribe(RxUtils.getObserver());
try {
    Thread.sleep(Integer.MAX_VALUE);
} catch (InterruptedException e1) {
    e1.printStackTrace();
}

  onSubscribe
  Thread:Thread[main,5,main]
  onNext:0
  Thread:Thread[RxComputationThreadPool-1,5,main]
  onNext:1
  Thread:Thread[RxComputationThreadPool-1,5,main]
  onNext:2
  Thread:Thread[RxComputationThreadPool-1,5,main]
  onNext:3
  Thread:Thread[RxComputationThreadPool-1,5,main]
  onNext:4
  Thread:Thread[RxComputationThreadPool-1,5,main]
  onComplete
  Thread:Thread[RxComputationThreadPool-1,5,main]

takeWhile

发射Observable发射的数据，直到一个指定的条件不成立

Observable.interval(1, TimeUnit.SECONDS).takeUntil(Observable.timer(10,TimeUnit.SECONDS))
        .subscribe(RxUtils.getObserver());
try {
    Thread.sleep(Integer.MAX_VALUE);
} catch (InterruptedException e1) {
    e1.printStackTrace();
}

onSubscribe
Thread:Thread[main,5,main]
onNext:0
Thread:Thread[RxComputationThreadPool-2,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-2,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-2,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-2,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-2,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-2,5,main]
onNext:6
Thread:Thread[RxComputationThreadPool-2,5,main]
onNext:7
Thread:Thread[RxComputationThreadPool-2,5,main]
onNext:8
Thread:Thread[RxComputationThreadPool-2,5,main]
onNext:9
Thread:Thread[RxComputationThreadPool-2,5,main]
onComplete
Thread:Thread[RxComputationThreadPool-1,5,main]