使用 RxJava retryWhen 运算符时 Kotlin 类型不匹配

Question

我正在尝试创建 Observable，它会在网络连接建立时重试。

我已经创建了主题：

 private val retrySubject = PublishSubject.create<Unit>()()

我是这样使用它的：

  private fun publishNetworkReconnection() {
    compositeDisposable?.add(
       connectionHelper.observeConnection()
        .subscribe {connected: Boolean
          if(connected){
          retrySubject.onNext(null)
          }
        }
     )
  }

然后我尝试在我的 retryWhen 运算符中使用它：

   val disposable =
    Flowable.interval(0, UPDATE_INTERVAL, TimeUnit.SECONDS, Schedulers.io())
        .onBackpressureDrop()
        .flatMapCompletable {
          revocationRepository.sync(event.id)
        }
        .retryWhen { retryHandler -> retryHandler.flatMap({ nothing -> retrySubject.asObservable() }) }
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe({ }, { Timber.e(it, "Unable to sync blacklist") })
    compositeDisposable?.add(disposable)
  }

这种情况下如何正确使用？

我在 Android Studio 中收到此错误：

类型不匹配。必需：发布者>!发现：可观察的!

Answer 1

问题是，您正在混合使用 RxJava 1 - Subject.asObservable 和 RxJava 2 - Flowable。

来自 RxJava 2 的retryWhen 的签名是：

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

所以 retryWhen 内的 lambda 应该返回扩展 Publisher 的东西。相反，您返回的是 Subject<Unit>，它被转换为 Observable<Void>，因为显然您的 Subject 来自 RxJava 1。而且 - 由于 RxJava 版本不同 - 显然它不会扩展 Publisher。

所以错误消息是正确的，retryWhen 期望来自 RxJava 2 的 Publisher，但您提供来自 RxJava 1 的 Obervable。您没有注意到不同的包，因为它们不包含在消息中。

主要问题是混合来自 RxJava 1 和 RxJava 2 的代码，这从来都不好。