来自不同线程的retryWhen

Question

我和RxJava有一种情况，我不太明白。我有一个可以观察到的环境，效果很好。

Observable.create(subscriber -> makeWebCall())
    .subscribeOn(Schedulers.newThread())
    .observeOn(AndroidScheduler.newThread())
    .retryWhen(observable ->
                observable.flatMap(throwable -> {
                    if (throwable instanceof UserRecoverableException) {
                        return Observable.from(someUIFlowObservable());
                    }
                    return Observable.error(throwable);
                }))
    .subscribe(response -> response.doSomething(), t -> throwError(t));

所以我进行了一个web调用，如果我得到了一个可恢复的异常，在UI线程中做了一些更改之后，我重新尝试了这个调用。

当然，这一切都是由于某些UI交互(即按钮单击)而发生的。还有一个可以观察到的链子，所以我修改了这个链子，让它也包括这个调用，就像这样：

Rx.click(someButton)
    .switchMap(o -> getSomeStringPreferenceObservable())
    .subscribeOn(AndroidSchedulers.mainThread())
    .observeOn(Schedulers.newThread())
    .switchMap(str -> Observable.create(subscriber - >makeWebCall()))
    .observeOn(AndroidScheduler.mainThread())
    .retryWhen(observable ->
                observable.flatMap(throwable -> {
                    if (throwable instanceof UserRecoverableException) {
                        return Observable.from(someUIFlowObservable());
                    }
                    return Observable.error(throwable);
                }))
    .subscribe(response -> response.doSomething(), t -> throwError(t));

现在，retryWhen不工作了。正如应该非常明显的那样，这两种方法之间最大的区别是subscribeOn方法。但我不太清楚发生了什么，我能做些什么才能得到理想的结果。

Answer 1

我认为它有效，你只需要再点击一下那个按钮。

原因是retryWhen重新订阅了整个链，但是Rx.click(someButton)没有发出任何新的内容(在重新订阅之后)，因此其余的不会被处理。

如果UserRecoverableException是由makeWebCall()发出的，那么将重试移动到可观察的：

.switchMap(str -> Observable.create(subscriber - > makeWebCall())
    .observeOn(Schedulers.newThread())
    .retryWhen(observable -> observable.flatMap(throwable -> {
      if (throwable instanceof UserRecoverableException) {
        return Observable.from(someUIFlowObservable())
            .subscribeOn(mainThread());
      }
      return Observable.error(throwable);
    }))
)

Ps:在新的AndroidScheduler.newThread()中没有RxAndroid :)

Ps 2:如果从某个活动或片段运行可观察到的内容，则不需要调用.subscribeOn(AndroidSchedulers.mainThread())。