使用 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 中收到此错误:

Type mismatch. Required: Publisher<< out (???..???) >>! Found: Observable< Void !>!

Answer 1

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

retryWhen 的签名|来自 RxJava 2 是:

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

所以里面的 lambda retryWhen应该返回扩展 Publisher 的东西.相反，您正在返回 Subject<Unit>转换为 Observable<Void> ，因为显然你的 Subject来自 RxJava 1。而且 - 由于 RxJava 版本差异 - 显然它没有扩展 Publisher .

所以错误信息是正确的， retryWhen预计 Publisher来自 RxJava 2 ，但你给 Obervable来自 RxJava 1 .您没有注意到不同的包，因为它们未包含在消息中。

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