我在使用rxjava时遇到了一个非常具体的问题或误解,希望有人能够帮我解决。
我正在运行rxjava 2.1.5,并且有以下代码片段:
public static void main(String[] args) {
final Observable<Object> observable = Observable.create(emitter -> {
// Code ...
});
observable.subscribeOn(Schedulers.io())
.retryWhen(error -> {
System.out.println("retryWhen");
return error.retry();
}).subscribe(next -> System.out.println("subscribeNext"),
error -> System.out.println("subscribeError"));
}
执行完此操作后,程序会输出:
retryWhen
Process finished with exit code 0
我的问题是:为什么在订阅Observable时立即调用retryWhen?这个Observable什么都不做。
我想要的是,当发射器上出现onError时,才调用retryWhen。我是否误解了rx的工作原理?
谢谢!
添加新代码片段:
public static void main(String[] args) throws InterruptedException {
final Observable<Object> observable = Observable.create(emitter -> {
emitter.onNext("next");
emitter.onComplete();
});
final CountDownLatch latch = new CountDownLatch(1);
observable.subscribeOn(Schedulers.io())
.doOnError(error -> System.out.println("doOnError: " + error.getMessage()))
.retryWhen(error -> {
System.out.println("retryWhen: " + error.toString());
return error.retry();
}).subscribe(next -> System.out.println("subscribeNext"),
error -> System.out.println("subscribeError"),
() -> latch.countDown());
latch.await();
}
调用了Emitter的onNext和complete方法,但是未调用DoOnError方法。输出结果如下:
retryWhen: io.reactivex.subjects.SerializedSubject@35fb3008 subscribeNext
进程以退出代码0结束。