我正在尝试创建一个Observable,当网络连接建立时将进行retryWhen操作。
我已经创建了一个subject:
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中遇到了这个错误:
类型不匹配。需要:Publisher << out (???..???) >>!找到:Observable < Void ! >!
Flowable
,另一方面有Subject.asObservable
,它在RxJava 1中。你同时使用这两个版本吗?即使编译此代码也很困难。 - michalbrz