java - 如何使用 RxJava 的方法 retryWhen 多次调用一个方法?
问题描述
我有一个执行网络请求并返回带有可变数据类列表的 Observable 的方法。有时此方法会因 403 错误而失败。我需要调用 YouTubeClient.getApiKey() 方法来获取新的 Api 密钥并重复对网络的请求。怎么做?我阅读了很多类似的主题,但没有找到可行的决定。
实用方法的这段代码,当我尝试调用 retryWhen() 方法时
private fun searchRequestWrapper(query: String): Observable<MutableList<Video>> {
return youTubeClient.searchRequest(
YouTubeClient.URL_SNIPPET,
YouTubeClient.MAX_RESULT, query,
YouTubeClient.API_KEY
)
.retryWhen { errors -> errors
.zipWith(Observable.range(1, 3)) { error, a ->
YouTubeClient.getApiKey()
error
}
}
.map {it.items}
}
Main 方法在下面调用自身内部的实用程序方法
fun fetchVideos(query:String) {
_networkState.set(NetworkState.LOADING)
Log.e("NetworkState", networkState.get()?.status.toString())
try {
compositeDisposable.add(
searchRequestWrapper(query)
.flatMapIterable {it}
.flatMap { video -> videoInfoWrapper(video.videoId).subscribeOn(Schedulers.io()) }
.toList()
.subscribeOn(Schedulers.io())
.subscribe({
Log.e("new videosId",it.toString())
downloadedVideosList.postValue(it)
_networkState.set(NetworkState.LOADED)
Log.e("NetworkState", networkState.get()?.status.toString())
_networkState.set(NetworkState.WAITING)
Log.e("NetworkState", networkState.get()?.status.toString())
},{
errorHandle(it)
}))
}
catch (e: Exception){
Log.e("fetchVideos",e.message)
}
}
解决方案
这个解决方案怎么样?
实际上你使用 retryWhen 是错误的。retryWhen 运算符只是指示是否应该发生重试。每次内部 observable 发出一个值时,都会开始重试。
在我的示例中,通过 onErrorResumNext 捕获了 403 异常,该异常捕获了 onError 发出并订阅了一个 fallback-observable。此运算符后的 retryWhen 确保其他错误将被重试 3 次。
仅供参考:每个运算符转换日期应该只适用于不可变数据,因此应该避免使用 MutableList。
class So65040953 {
@Test
fun main() {
val client = YouTubeClientInterceptor(YouTubeClientStub())
val scheduler = TestScheduler()
val test = client.searchRequest("fail")
// when 403 -> request key and call search again...
// else propagate error
.onErrorResumeNext {
when (it.message) {
"403" -> client.requestApiKey()
.flatMapObservable { key -> client.searchRequest("good") }
else -> Observable.error(it)
}
}
// when some error other than 403 happens... retry ever second for three times....
// error could be thrown by #searchRequest or #requestApiKey
.retryWhen { throwableObservable ->
throwableObservable.take(3).delay(1L, TimeUnit.SECONDS, scheduler)
}.test()
scheduler.advanceTimeBy(10, TimeUnit.SECONDS)
assertThat(client.requestApiKey.get()).isEqualTo(1)
assertThat(client.searchRequestCount.get()).isEqualTo(2)
test.assertNoErrors()
.assertValue(listOf(Video("42")))
}
}
internal interface YouTubeClient {
fun searchRequest(query: String): Observable<List<Video>>
fun requestApiKey(): Single<String>
}
internal class YouTubeClientInterceptor(private val client: YouTubeClient) : YouTubeClient {
internal val searchRequestCount = AtomicInteger(0)
internal val requestApiKey = AtomicInteger(0)
override fun searchRequest(query: String): Observable<List<Video>> {
searchRequestCount.incrementAndGet()
return client.searchRequest(query)
}
override fun requestApiKey(): Single<String> {
requestApiKey.incrementAndGet()
return client.requestApiKey()
}
}
internal class YouTubeClientStub : YouTubeClient {
override fun searchRequest(query: String): Observable<List<Video>> {
println("query: $query")
return when (query) {
"fail" -> Observable.error<List<Video>>(RuntimeException("403"))
"good" -> Observable.just(listOf(Video("42")))
else -> Observable.empty()
}
}
override fun requestApiKey(): Single<String> {
println("requestApiKey")
return Single.just("1-1-1-1")
}
}
internal data class Video(private val videoId: String)
推荐阅读
- concurrency - 如何在 rust 中执行线程安全 IO 和缓存到文件?
- ssis - 如何在 C# 中修改数据表
- c# - 我是否需要创建一个存储过程来检查主机名是否存在?
- php - 我已经尝试了堆栈溢出的所有解决方案然后我也收到语法错误,意外的'endwhile'(T_ENDWHILE)
- reactjs - 我想在本机反应中用按钮制作一个圆圈
- assembly - 使用汇编语言的阶乘程序,用户需要输入一个整数。但是,我得到了非法指令
- ios - 保存的透明图像在 Swift 中返回黑色背景
- python-3.x - ModuleNotFoundError:没有名为“帐户”的模块
- python - 将 matplotlib savefig .png 传输到 dest 文件夹
- python - 如何在数据框中仅选择索引列?