1.all操作符
判断被观察者所有的发送事件，是否满足自定义的条件。
如果全部满足，则观察者接受到的事件类型为true否则为false
/**
* all:判断所有的事件是否都满足条件
* 都满足 则 发送的事件为true
* 否则发送的事件为false
*/
private void all() {
Observable.just(1,2,3,4,5).all(new Predicate<Integer>() {
@Override
public boolean test(Integer integer) throws Exception {
//自定义条件
return integer<3;
}
}).subscribe(new Consumer<Boolean>() {
@Override
public void accept(Boolean aBoolean) throws Exception {
Log.e(TAG, "accept: "+aBoolean );
}
});
}
输出结果：
由于有4和5大于3，所以最终观察者接受到的事件为false
com.mujin.keji.rxjava_01 E/MainActivity: accept: false
2.takeWhile
发送满足条件的事件
/**
* 发送满足条件的事件
*/
private void takeWhile() {
Observable.just(1,2,3,4,5).takeWhile(new Predicate<Integer>() {
@Override
public boolean test(Integer integer) throws Exception {
return integer<3;
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "accept: "+integer );
}
});
}
输出结果：
04-09 15:18:15.609 25910-25910/com.mujin.keji.rxjava_01 E/MainActivity: accept: 1
04-09 15:18:15.609 25910-25910/com.mujin.keji.rxjava_01 E/MainActivity: accept: 2
3.skipeWhile
跳过满足条件的事件
/**
* 跳过满足条件的事件
*
*/
private void skipWhile() {
Observable.just(1,2,3,4).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 {
Log.e(TAG, "accept: "+ integer);
}
});
}
输出结果：
04-09 15:21:08.042 27151-27151/com.mujin.keji.rxjava_01 E/MainActivity: accept: 2
04-09 15:21:08.042 27151-27151/com.mujin.keji.rxjava_01 E/MainActivity: accept: 3
04-09 15:21:08.042 27151-27151/com.mujin.keji.rxjava_01 E/MainActivity: accept: 4
4.takeUnitil
执行到某个条件之后
停止发送事件
/**
* takeUnitil执行到某个条件之后
* 停止发送事件
*/
private void takeUnitil() {
Observable.just(1,2,3,4,5).takeUntil(new Predicate<Integer>() {
@Override
public boolean test(Integer integer) throws Exception {
//当执行完大于等于三之后的事件停止发送
return integer>=3;
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "accept: "+integer );
}
});
}
输出结果:
04-09 15:24:53.973 29195-29195/com.mujin.keji.rxjava_01 E/MainActivity: accept: 1
04-09 15:24:53.973 29195-29195/com.mujin.keji.rxjava_01 E/MainActivity: accept: 2
04-09 15:24:53.974 29195-29195/com.mujin.keji.rxjava_01 E/MainActivity: accept: 3
5.sequenceEqual
判断两个观察者发送的事件是否相同
/**
* 判断两个观察者发送的事件是否相同
*/
private void sequenceEqual() {
Observable.sequenceEqual( Observable.just(1,2,3,4),
Observable.just(2,3,3,4)).subscribe(new Consumer<Boolean>() {
@Override
public void accept(Boolean aBoolean) throws Exception {
Log.e(TAG, "accept: "+aBoolean );
}
});
}
输出结果:
04-09 15:33:45.583 31225-31225/com.mujin.keji.rxjava_01 E/MainActivity: accept: false
6.contains
判断发送的事件是否包含某个事件
/**
* 判断发送的事件是否包含某个事件
*/
private void contains() {
Observable.just(1,2,3).contains(1).subscribe(new Consumer<Boolean>() {
@Override
public void accept(Boolean aBoolean) throws Exception {
Log.e(TAG, "accept: "+aBoolean );
}
});
}
输出结果：
04-09 15:36:19.785 32357-32357/com.mujin.keji.rxjava_01 E/MainActivity: accept: true
7.isEmpty
判断是否为空事件
/**
* 判断事件是否为空
*/
private void isEmpty() {
Observable.just(1,2,3).isEmpty().subscribe(new Consumer<Boolean>() {
@Override
public void accept(Boolean aBoolean) throws Exception {
Log.e(TAG, "accept: "+aBoolean );
}
});
}
输出结果：
04-09 15:39:07.105 1136-1136/com.mujin.keji.rxjava_01 E/MainActivity: accept: false
8.defaultIfEmpty
在不发送任何事件，只发送一个Complete
事件的前提下，给一个默认事件
/**
* 在不发送任何事件，只发送一个Complete
* 事件的前提下，给一个默认值
*/
private void defaultIfEmpty() {
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onComplete();
}
}).defaultIfEmpty(10).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "accept: "+integer );
}
});
}
输出结果：
04-09 15:42:22.335 2856-2856/com.mujin.keji.rxjava_01 E/MainActivity: accept: 10
