Start obserable chain from Single.onErrorResumeNext in Rxjava2

Question

loadSingle return Single object, if it fails I want to call getObservable(rsList) which return Observable. I am trying with onErrorResumeNext but it needs Single object. How can I call getObservable(rsList) on failure of loadSingle() ?

Thanks in advance!!

                    repo.loadSingle()
                            .subscribeOn(Schedulers.io())
                            
                            .onErrorResumeNext {
                                repo.getObservable(rsList)
                            }
                            .flatMapObservable {
                                if (it != null && it.status == Status.SUCCESS) {
                                    upRsList(it.data)
                                }
                                repo.getObservable(rsList)
                            }
                          
                            ({ //observable success
                             }, {
                               //observable error 
                            })

Api interface

interface HomeApi{
    fun getSingel():Single<List<String>>
    fun getObservable():Observable<HomeResponse>
}

Answer 1

Dependencies

testImplementation("org.junit.jupiter:junit-jupiter-api:5.6.2")
testRuntimeOnly("org.junit.platform:junit-platform-launcher:1.6.2")
testRuntimeOnly("org.junit.jupiter:junit-jupiter-engine:5.6.2")
testRuntimeOnly("org.junit.vintage:junit-vintage-engine:5.6.2")
implementation "io.reactivex.rxjava3:rxjava:3.0.4"
implementation "io.reactivex.rxjava3:rxkotlin:3.0.0"

Required classes

internal interface Repo {
    fun loadSingle(): Single<Result<List<String>>>

    fun getObservable(list: List<String>): Observable<String>
}

internal class RepoImpl : Repo {
    override fun loadSingle(): Single<Result<List<String>>> {
        return Single.error(RuntimeException("fail"))
    }

    override fun getObservable(list: List<String>): Observable<String> {
        if (list === emptyList<String>()) {
            return Observable.just("42")
        }
        return Observable.just("success")
    }
}

internal sealed class Result<T> {
    data class Success<T>(val value: T) : Result<T>()

    data class Failure<T>(private val failure: Throwable) : Result<T>()
}

Test

Wrap the error via #onErrorReturn into a default value, and handle the result accordingly.

class So64751341 {
    @Test
    fun `64751341`() {
        val repo: Repo = RepoImpl()
        val testScheduler = TestScheduler()

        val flatMapObservable = repo.loadSingle()
            .subscribeOn(testScheduler)
            .onErrorReturn { failure -> Result.Failure(failure) }
            .flatMapObservable { result ->
                when (result) {
                    is Result.Success -> repo.getObservable(result.value)
                    is Result.Failure -> repo.getObservable(emptyList())
                }
            }


        val test = flatMapObservable.test()

        testScheduler.triggerActions()

        test // return default value 42 onError
            .assertValue("42")
    }
}

Repo#loadSingle() throws exception synchronously

internal class RepoExceptionImpl : Repo {
    override fun loadSingle(): Single<Result<List<String>>> {
        throw java.lang.RuntimeException("whatever")
    }

    override fun getObservable(list: List<String>): Observable<String> {
        if (list === emptyList<String>()) {
            return Observable.just("42")
        }
        return Observable.just("success")
    }
}

Test

Repo#loadSingle must be wrapped with Single#defer. Single#defer will catch the exception and emit it as #onError to the subscriber, which in turn will be handled by #onErrorReturn

@Test
fun `64751341_exception`() {
    val repo: Repo = RepoExceptionImpl()
    val testScheduler = TestScheduler()

    val flatMapObservable = Single.defer {
        repo.loadSingle()
    }
        .subscribeOn(testScheduler)
        .onErrorReturn { failure -> Result.Failure(failure) }
        .flatMapObservable { result ->
            when (result) {
                is Result.Success -> repo.getObservable(result.value)
                is Result.Failure -> repo.getObservable(emptyList())
            }
        }

    val test = flatMapObservable.test()

    testScheduler.triggerActions()

    test // return default value 42 onError
        .assertValue("42")
}