Как разорвать цепочку RxJava при ошибке Result?

Question

Я унаследовал эту кодовую базу, которая использует RxJava2 и kotlin с довольно своеобразным шаблоном Result для вызовов API. т.е. все вызовы API возвращают Singles с объектом Result (который является запечатанным классом типов Success и Error, как показано ниже), т.е.

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

Теперь я пытаюсь связать воедино несколько API-вызовов, но нужно завершить цепочку в первом Result.Error и продолжить, если нет.

Единственный способ, которым я могу придумать, - это сжать все синглы, а затем использовать функцию молнии, которая проверяет тип каждого параметра и возвращает Result.Error() с первой ошибкой, с которой он сталкивается. то есть что-то вроде

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

который работает, но выглядит действительно странно (и ощущается как запах кода с помощью этого метода с огромной задницей на молнии). Также не позволяет мне связывать что-либо еще после последнего метода (который проверяет, является ли Результат успехом / ошибкой).

Я чувствую, что было бы намного удобнее читать цепочку этих вызовов и завершать работу при первой ошибке, но я не знаю достаточно Rx для этого. Есть оператор или подход, который может помочь сделать это лучше?

Answer 1

Вы можете получить оригинал Single поведение путем изменения того, что ваша кодовая база уже делает.

Создайте преобразователь, который будет извлекать данные из вызова API или выдавать ошибку при ошибке. Первая ошибка прекратится 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());
    });
}

Используйте это с repo.makeCall(arg).compose(transform())

Надеюсь, поможет.

Answer 2

Из коробки RxJava "прервал бы при первой ошибке", потому что Observable и Single (что сродни Task/Future/Promise) обладает "монадическими качествами". Но, как Result<*, *> явным образом обрабатывает ошибки на пути "успеха", чтобы избежать прерывания потока, мы могли бы рассмотреть другой путь, чем разрешение Rx перейти к событиям терминала - потому что существующий код ожидает, что он будет на пути успеха. Терминальные события должны быть для исключений "конец света", а не для тех, которые мы ожидаем и можем обработать.

---

У меня были некоторые идеи, но я думаю, что единственное, что вы можете сделать, - это уменьшить количество строк, необходимых для этого, вместо того, чтобы вычеркнуть его.

Технически мы пытаемся повторно реализовать Either<E, T> Монада от Arrow, но, зная это, мы можем уменьшить количество строк с помощью некоторых хитростей:

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

Что вы думаете об этом блоке кода:

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

Я объединяю результаты в list, затем сопоставьте его с ожидаемым результатом. Это намного легче читать.