在重试时修改源Observable - RxJava

Question

如何在重试时更新源可观察对象？

List<String> ids = new ArrayList<>(); // A,B,C
Observable.from(ids)
          .retryWhen(errors -> {
                    return errors
                    .zipWith(Observable.range(0, 1), (n, i) -> i)
                    .flatMap(retryCount -> Observable.timer((long) Math.pow(2, retryCount), TimeUnit.MINUTES));

           })
           .subscribe(....);

现在，如果我想传递一些其他值而不是作为id的//A、B、C，我该怎么做？这样做是否正确？

- Bharath

Answer 1

使用defer。这将允许重新计算ids：

Observable.defer(() -> {
    List<String> ids = // compute this somehow
    return Observable.from(ids);
}).retryWhen(...

Documentation on the defer operator

- drhr

---

我只有一个订阅者。defer为每个观察者创建一个新的可观察对象。你不认为onErrorResumeNext是更好的选择吗？ - Bharath

你能澄清一下你的意思吗？订阅回调本身就是观察者 - 它从上游接收传入的项。因此，“可观察”这个术语指的是你正在观察的事物 - 它正在产生可以被观察者观察的项。多个观察者可能意味着你正在进行多播，但在这个例子中你并没有这样做。 - drhr

Answer 2

onErrorResumeNext 可以使用。您可能需要一些额外的逻辑来匹配您的用例。错误处理运算符的文档 在这里。

List<String> ids = new ArrayList<>(); // A,B,C
List<String> ids2 = new ArrayList<>(); // D,E,F
Observable.from(ids)
        .onErrorResumeNext(throwable -> {
            return Observable.from(ids2);
        });

- ehehhh

---

当 retryWhen 计时器到期时，将调用 onErrorResumeNext 吗？ - Bharath

目前，当第一个可观察对象发出错误后，它将立即被调用，您必须自己添加定时器逻辑。 - ehehhh

我已经在retryWhen中有一个计时器逻辑。但是，我认为计时器逻辑应该作为onErrorResumeNext的一部分吗？ - Bharath

是的，在这种情况下，计时器逻辑必须放在 onErrorResumeNext 中。 - ehehhh