RxJava Filter on Error

Question

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).

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.

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.

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)
}

Answer 1

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()
}