AndThen — RxKotlin chaining sources improvement

In this article, I show a nice way to chain operations and handle errors.

I recently faced a use case that I solved easily with Rx. I want to execute two different actions one after the other but still differentiate the errors origin.

Chaining Single operations with andThen

The andThen method is available on a Completable object. “It returns a Single which will subscribe to this Completable and once that is completed then will subscribe to the next SingleSource”, but I have two Single so I use toCompletable to return a Completable.

val firstSingle = firstOperation()
val secondSingle = secondOperation()

firstSingle.toCompletable()
        .andThen(secondSingle)
        .subscribeBy({}, {})

Note that in this case, I don’t really care about firstOperation’s result. I just need to know that the operation has finished.
Tips with Kotlin : It is possible to add names to call arguments.

.subscribeBy(onSuccess = {}, onError = {})

With this code, I can deal with my Throwable in onError{}. But in the case that firstSingle and secondSingle potentialy return the same error, I’m not able to differentiate if the error is thrown by firstSingle or secondSingle.

Differentiate errors origin with onErrorResumeNext

The goal is to add an additional information on the Throwable. In order to do this, I create a Usecase enum and a CustomError which extends Throwable.

enum class UseCase { FIRST_OPERATION, SECOND_OPERATION }

class CustomError(
        val useCase: UseCase,
        override val cause: Throwable) : Throwable()

Now for each operation, I intercept the Throwable emissions and emit a CustomError instead. The right way to do this is to use onErrorResumeNext. “It instructs a Single to pass control to another Single rather than invoking SingleObserver#onError(Throwable) if it encounters an error.” Just make this second Single call SingleObserver#error(CustomError).

val firstSingle = firstOperation()
       .onErrorResumeNext { Single.error(CustomError(useCase = FIRST_OPERATION, cause = it.cause)) }

val secondSingle = secondOperation()
       .onErrorResumeNext { Single.error(CustomError(useCase = SECOND_OPERATION, cause = it.cause)) }

firstSingle.toCompletable()
        .andThen(secondSingle)
        .subscribeBy(
                onSuccess = { },
                onError = { it as? CustomError }
        )

Finally in onError{ }, I can cast the Throwable into a CustomError, which allows to differentiate use cases. For example :

onError = {
    (it as? CustomError)?.let {
        when (it.useCase) {
            FIRST_OPERATION -> askForHelp(it.message)
            SECOND_OPERATION -> callTheCops(it.message)
        }
    }
}

Thanks Kotlin !