rx-java2 - RetryWhen() 延迟输入错误

Question

我正在关注 this tutorial然后添加这一行:

.retryWhen(errors -> errors.flatMap(error -> Observable.timer(30, TimeUnit.SECONDS)))

到我的Transformer，但它触发了一个编译错误:

error: incompatible types: cannot infer type-variable(s) R (argument mismatch; bad return type in lambda expression Observable cannot be converted to Publisher) where R,T are type-variables: R extends Object declared in method flatMap(Function>) T extends Object declared in class Flowable

error -> Observable.timer(30, TimeUnit.SECONDS) 下面的波浪线表示:

no instance(s) of type variable(s) R exist so that Observable conforms to Publisher

我做错了什么？缺少什么才能像示例中那样工作？

Answer 1

What am I doing wrong?

请检查运算符(operator)的签名，以便您使用正确的类型:https://github.com/ReactiveX/RxJava#base-class-vs-base-type

JavaDoc :

public final Single<T> retryWhen(
    Function<? super Flowable<Throwable>,? extends Publisher<?>> handler)

Because this tutorial uses Observable.timer with no issues.

该教程早于 RxJava 2。事实上，上面链接的 Javadoc 包含一个带有 Flowable.timer() 的示例:

Single.timer(1, TimeUnit.SECONDS)
.doOnSubscribe(s -> System.out.println("subscribing"))
.map(v -> { throw new RuntimeException(); })
.retryWhen(errors -> {
    AtomicInteger counter = new AtomicInteger();
    return errors
              .takeWhile(e -> counter.getAndIncrement() != 3)
              .flatMap(e -> {
                  System.out.println("delay retry by " + counter.get() + " second(s)");
//                vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv
                  return Flowable.timer(counter.get(), TimeUnit.SECONDS);
//                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
              });
})
.blockingGet();

is this something only related to Single

retryWhen 和 repeatWhen 在设计上使用 Publisher 作为重做信号，这样我们就可以利用背压在一个时间。使用 Observable，处理程序有可能一次简单地转储大量信号，运算符(operator)可能会出现意外行为。