RxJava: Execute second observables only if first one throws an error and repeat from the first

Question

I am using retorift to hit getAricle api and get list of articles related to the user. getArticle api will throw error if token passed is expired if so then I have to call refreshToken api to get new token then again I have to call the getArticle api

 ApiController.createRx().getArticle(token)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({ response -> toast(response.body().url) }, { e ->
                println(e.printStackTrace())
                if(e is HttpException && e.code() in  arrayOf(401,403)){                      
                   //Here I want to call refresh tolken api
                   toast("Auth error")
                }
                else
                   toast(R.string.something_went_wrong)
            })

Edit

Even though given answers showed some direction but those are not a direct answer to my question. This is how solved it but I feel this can be refactored into much better code

ApiController.createRx().getArticle(Preference.getToken())
            .flatMap { value ->
                if (value.code() in arrayOf(403, 401)) {
                    ApiController.refreshToken()
                    ApiController.createRx().getArticle(Preference.getToken())
                } else Observable.just(value)
            }
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({ response -> println("Success") }, { e ->
                e.printStackTrace()
                toast(R.string.something_went_wrong)
            })



fun refreshToken() {
        val token:String?=ApiController.createRx().refreshToken(Preferences.getRefreshToken()).blockingFirst()?.body()?.token
        if (token != null) Preferences.setAuthToken(token)
    }

EDIT

I refactored my code to little more cleaner version

Observable.defer { ApiController.createRx().getArticle(Preferences.getToken()) }
            .flatMap {
                if (it.code() in arrayOf(401, 403)) {
                    ApiController.refreshToken()
                    Observable.error(Throwable())
                } else Observable.just(it)
            }
            .retry(1)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({println("Success") }, {
              it.printStackTrace()
              toast(R.string.something_went_wrong)
            })



 fun refreshToken() {
        var token: String? = null
        try {
            token = createRx().refreshToken(Preferences.getRefreshToken()).blockingFirst().body()!!.token
        } catch (e: Exception) {
            throw e
        }
        println("saving token")
        if (token != null) Preferences.setAuthToken(token)
    }

EDIT

Please check my answer for the final refactored code

Answer 1

I solved my problem after reading more about RxJava and this is how I implemented it. First of all will retrofit throw 4xx error to onError or onNext\onSuccess depends on how we define it. Ex:

@GET("content") fun getArticle(@Header("Authorization") token: String):Single<Article>

this will throw all the 4xx errors to onError and instead of Single<Article> if you define it as Single<Response<Article>> then all the response from server including 4xx will go to onNext\onSuccess

Single.defer { ApiController.createRx().getArticle(Preferences.getAuthToken())}
                .doOnError {
                    if (it is HttpException && it.code() == 401)
                        ApiController.refreshToken()
                }
                .retry { attempts, error -> attempts < 3 && error is HttpException && error.code() == 401 }
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe({println("Success") }, {
                  it.printStackTrace()
                  toast(R.string.something_went_wrong)
                })

I am using defer as a wrapper around my actual Observable because I want to recreate the article fetch observable on retry after token refresh because I want Preferences.getAuthToken() to be called again as my refresh token code stores newly fetched token in preference.

retry returns true if the HttpException is 401 and not attempted retry more than 2 times