repeatWhen可以实现重订阅功能,而触发重订阅两个关键因素:
- Obervable完成一次订阅,就是Observable调用onCompleted
- 当Observable调用onCompleted就会进入到repeatWhen方法里面,是否要触发重订阅,就需要通过repeatWhen的Function方法所返回的ObservableSource确定,如果返回的是onNext则触发重订阅,而返回的是onCompleted/onError则不会触发重订阅
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onComplete();
}
}).doOnComplete(new Action() {
@Override
public void run() throws Exception {
Log.d(TAG, "触发重订阅");
}
}).repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
private int n = 0;
@Override
public ObservableSource<?> apply(Observable<Object> objectObservable) throws Exception {
return objectObservable.flatMap(new Function<Object, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(Object o) throws Exception {
if (n != 3) {
n++;
return Observable.timer(3, TimeUnit.SECONDS);
} else {
return Observable.empty();
}
}
});
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
Log.d(TAG, "onNext: " + s);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: " + e);
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: ");
}
});
运行结果:
08-02 10:26:13.187 4992-4992/com.zhen.rxjavademo D/MainActivity-vv: 触发重订阅
08-02 10:26:16.196 4992-5013/com.zhen.rxjavademo D/MainActivity-vv: 触发重订阅
08-02 10:26:19.204 4992-5019/com.zhen.rxjavademo D/MainActivity-vv: 触发重订阅
08-02 10:26:22.206 4992-5020/com.zhen.rxjavademo D/MainActivity-vv: 触发重订阅
需要注意的是repeatWhen的objectObservable必须处理,这里使用flathMap进行处理,让它延时发出onNext,这里onNext发出什么数据都不重要,它只是仅仅用来处理重订阅的通知,如果发出的是onCompleted/onError,则不会触发重订阅
doOnCompleted 当Observable调用onCompleted就会被触发
Observable.time 延迟一段时间后,发送onNext,数据为0
Observable.empty 发送onCompleted,Observer不会回调onCompleted
Observable.error 发送onError,Observer的onError会被回调,并接收错误信息
repeatWhen可以用于延时轮询,在doOnCompletetd进行操作