RxJava过滤错误 [英] RxJava Filter on Error

问题描述

该问题与此问题松散相关，但是有没有答案.鲍勃·达尔格利什(Bob Dalgleish)的答案很接近，但不支持来自Single的潜在错误(我认为OP也是想要的).

This question is loosely related to this question, but there were no answers. The answer from Bob Dalgleish is close, but doesn't support the potential error coming from a Single (which I think that OP actually wanted as well).

我基本上是在寻找一种根据错误进行过滤"的方法-但是当查找基于RX时，这并不存在.我正在尝试获取值列表，通过查找运行它们，并跳过任何返回查找失败(可抛出)的结果.我在弄清楚如何以被动方式完成此操作时遇到了麻烦.

I'm basically looking for a way to "filter on error" - but don't think this exists when the lookup is RX based. I am trying to take a list of values, run them through a lookup, and skip any result that returns a lookup failure (throwable). I'm having trouble figuring out how to accomplish this in a reactive fashion.

我尝试了各种组合形式的错误处理运算符与映射.过滤器仅适用于原始值-至少我无法弄清楚如何使用它来支持我想做的事情.

I've tried various forms of error handling operators combined with mapping. Filter only works for raw values - or at least I couldn't figure out how to use it to support what I'd like to do.

在我的用例中，我迭代了一个ID列表，从远程服务请求每个ID的数据.如果服务返回404，则该项目不再存在.我应该从本地数据库中删除不存在的项目，然后继续处理ID.该流应返回查找值的列表.

In my use case, I iterate a list of IDs, requesting data for each from a remote service. If the service returns 404, then the item doesn't exist anymore. I should remove non-existing items from the local database and continue processing IDs. The stream should return the list of looked up values.

这是一个宽松的例子.如何编写getStream()以便canFilterOnError通过?

Here is a loose example. How do I write getStream() so that canFilterOnError passes?

import io.reactivex.Single
import io.reactivex.schedulers.Schedulers
import org.junit.Test

class SkipExceptionTest {

    private val data: Map<Int, String> = mapOf(
            Pair(1, "one"),
            Pair(2, "two"),
            Pair(4, "four"),
            Pair(5, "five")
    )

    @Test
    fun canFilterOnError() {

        getStream(listOf(1, 2, 3, 4, 5))
                .subscribeOn(Schedulers.trampoline())
                .observeOn(Schedulers.trampoline())
                .test()
                .assertComplete()
                .assertNoErrors()
                .assertValueCount(1)
                .assertValue {
                    it == listOf(
                            "one", "two", "four", "five"
                    )
                }
    }

    fun getStream(list: List<Int>): Single<List<String>> {
        // for each item in the list
        // get it's value via getValue()
        // if a call to getValue() results in a NotFoundException, skip that value and continue
        // mutate the results using mutate()

        TODO("not implemented")
    }

    fun getValue(id: Int): Single<String> {
        return Single.fromCallable {
            val value: String? = data[id]
            if (value != null) {
                data[id]
            } else {
                throw NotFoundException("dat with id $id does not exist")
            }
        }
    }

    class NotFoundException(message: String) : Exception(message)
}

推荐答案

我最终将getValue()映射到Optional<String>，然后在其上调用onErrorResumeNext()并返回Single.error()或Single.just(Optional.empty()).从那里，主流可以过滤掉空的Optional.

I ended up mapping getValue() to Optional<String>, then calling onErrorResumeNext() on that and either returning Single.error() or Single.just(Optional.empty()). From there, the main stream could filter out the empty Optional.

private fun getStream(list: List<Int>): Single<List<String>> {
    return Observable.fromIterable(list)
            .flatMapSingle {
                getValue(it)
                        .map {
                            Optional.of(it)
                        }
                        .onErrorResumeNext {
                            when (it) {
                                is NotFoundException -> Single.just(Optional.empty())
                                else -> Single.error(it)
                            }
                        }
            }
            .filter { it.isPresent }
            .map { it.get() }
            .toList()
}

这篇关于RxJava过滤错误的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！