How to break RxJava chain on error Result?

Question

I've inherited this codebase which uses RxJava2 and kotlin with a rather peculiar Result pattern for API calls. i.e. all API calls return Singles with a Result object (which is a sealed class of Success and Error types as shown below) i.e.

sealed class Result<T, E> {
    data class Success<T, E>(
            val data: T
    ): Result<T, E>()

    data class Error<T, E>(
            val error: E
    ): Result<T, E>()
}

Now I'm trying to chain together a bunch of API calls of these but need to terminate the chain on the first Result.Error in it and continue if not.

The only way I can think of is to zip all of the Singles and then have a zipper function that checks the type of each parameter and returns a Result.Error() with the first error it encounters. i.e. something like,

Singles.zip(
    repo1.makeCall1(arg),
    repo1.makeCall2(arg2),
    repo2.makeCall1(arg3)
) { result1, result2, result3 ->
    val data1 = when (result1) {
        is Result.Error -> return@zip Result.Error(result1.error)
        is Result.Success -> result1.data
    }
    val data2 = when (result2) {
        is Result.Error -> return@zip Result.Error(result2.error)
        is Result.Success -> result2.data
    }
    val data3 = when (result3) {
        is Result.Error -> return@zip Result.Error(result3.error)
        is Result.Success -> result3.data
    }

    return@zip Result.Success(MergedData(data1, data2, data3))
}

which works but looks really weird (and feels like a code smell with this huge ass zipper method). Also does not allow me to chain anything more after the last method (that checks if the Result is a Success / Error).

I feel it would be a lot more readable to be able to chain these calls and terminate on the first error but I don't know enough Rx to do this. Is there an operator or an approach that could help make this better?

Answer 1

You can get original Single behavior by reversing what your codebase already does.

Create transformer which will extract data from api call or throw error on error. First error will terminate zip.

public <T, E extends Throwable> SingleTransformer<Result<T, E>, T> transform() {
    return source -> source.flatMap(result -> {
        if (result instanceof Result.Success) 
            return Single.just(((Success<T, E>) result).getData());
          else
            return Single.error(((Error<T, E>) result).getError());
    });
}

Use it with repo.makeCall(arg).compose(transform())

Hope it helps.

Answer 2

Out of the box, RxJava would "abort on the first error" because Observable and Single (which is akin to Task/Future/Promise) has "monadic qualities". But as Result<*, *> explicitly makes errors be handled on the "success" path to avoid aborting the stream, we might want to consider a different route than letting Rx go to terminal events - because the existing code expects it to be on the success path. Terminal events should be for "the world is ending" exceptions, not ones we actually expect and can handle.

---

I had some ideas but I think the only thing you can do is reduce the number of lines it takes to do this instead of flat-out removing it.

Technically we are trying to re-implement the Either<E, T> monad here from Arrow, but knowing that, we can reduce the number of lines with some tricks:

sealed class Result<T, E>(
    open val error: E? = null,
    open val data: T? = null
) {
    data class Success<T>(
        override val data: T
    ): Result<T, Nothing?>()

    data class Error<E>(
        override val error: E
    ): Result<Nothing?, E>()
}

fun <E> E.wrapWithError(): Result.Error<E> = Result.Error(this) // similar to `Either.asLeft()`
fun <T> T.wrapWithSuccess(): Result.Success<T> = Result.Success(this)  // similar to `Either.asRight()`

fun blah() {
    Singles.zip(
        repo1.makeCall1(arg),
        repo1.makeCall2(arg2),
        repo2.makeCall1(arg3)
    ) { result1, result2, result3 ->
        val data1 = result1.data ?: return@zip result1.error.wrapWithError()
        val data2 = result2.data ?: return@zip result2.error.wrapWithError()
        val data3 = result3.data ?: return@zip result3.error.wrapWithError()

        Result.Success(MergedData(data1, data2, data3))
    }
}

Answer 3

What do you think about this code block:

Single.zip(
        Single.just(Result.Error(error = 9)),
        Single.just(Result.Success(data = 10)),
        Single.just(Result.Success(data = 11)),
        Function3<Result<Int, Int>, Result<Int, Int>, Result<Int, Int>, List<Result<Int, Int>>> { t1, t2, t3 ->
            mutableListOf(t1, t2, t3)
        })
        .map { list ->
            list.forEach {
                if (it is Result.Error){
                    return@map it
                }
            }
            return@map Result
        } // or do more chain here.
        .subscribe()

I combine the results into a list, then map it to your expected result. It's much easier to read.