Q:

New single after error

i develop my own download user data flow, and have problem. Buyt first some code, method explanation:
loginManager.getAccessToken() //Returns Single<String> where string is my access token
loginManager.getRefreshToken() //Returns Single<String> where string is my refresh token
loginManager.downloadUserAccountData(String accessToken) //Returns Single<String> where string is my json response
loginManager.refreshAccessToken(String refreshToken) //Returns Single<String> where string is new access token
userManager.parseUserJson(String json) //Returns Single<User> where User is my user model
userManager.storeUser(User user) //Returns Completable

Currently I doning something like that:
Completable getAndStoreUserData = loginManager.getAccessToken()
        .flatMap(loginManager::downloadUserAccountData)
        .flatMap(userManager::parseUserJson)
        .flatMapCompletable(userManager::storeUser);

But this does not support, refresh access token and retry download. Server throw 403 when my accessToken is not current, so i return error in that case in my function downloadUserAccountData
Now i have tried to use doOnError() and onErrorReturn() but both of this operartors don't met my requirements. This is because doOnError() allow only to e.g. display Log and dont return anything. onErrorReturn() lets me return only value not a Single<Value>.
This is pseudocode i trying to achieve:
Completable getAndStoreUserData = loginManager.getAccessToken()
        .flatMap(loginManager::downloadUserAccountData)
        .onErrorReturn(error -> loginManager.getRefreshToken()
                .flatMap(loginManager::refreshAccessToken)
                .flatMap(loginManager::downloadUserAccountData)
        .flatMap(userManager::parseUserJson)
        .flatMapCompletable(userManager::storeUser);

Any examples appreciated :)

A:

I finally managed, how to do it. As @Cochi proposed i have used retryWhen() function. This is very simple example based on String for anyone else:
String body = null;

@Override
protected void onStart() {
    super.onStart();

    Single<String> stringSingle = Single.create(source -> {
                if(body != null) {
                    source.onSuccess(body);
                }
                else {
                    source.onError(new Exception("403"));
                }
            })
            .retryWhen(errors -> errors.flatMap(error -> {
                if(error.getMessage().contains("403")) {
                    Throwable throwable = refresh().blockingGet();
                    if(throwable == null) {
                        return Flowable.just(new Object());
                    }
                    return Flowable.error(throwable);
                }
                return Flowable.error(error);
            }))
            .flatMap(obj -> Single.just(obj.toString()));
}

public Completable refresh() {
    return Completable.create(source -> {
        try {
            body = "Success";
            source.onComplete();
        }
        catch (Exception e) {
            source.onError(e);
        }
    });
}

And there is full example, using my code:
Completable getAndStoreUserData() {}
    Completable getAndStoreUserData = loginManager.getAccessToken()
            .flatMap(loginManager::downloadUserAccountData)
            .retryWhen(errors -> errors.flatmap(this::retryConditionChecker))
            .flatMapCompletable(userManager::storeUser);

    return getAndStoreUserData
}

Publisher<?> retryConditionChecker(Throwable error) {
    if(error.getMessage().contains("403")) {

        Throwable throwable = loginManager.getRefreshToken()
                .flatMap(loginManager::refreshAccessToken)
                .flatMapCompletable(loginManager::replaceAccessToken)
                .blockingGet();

        if(throwable == null) {
            return Flowable.just(new Object());
        }
        return Flowable.error(throwable);
    }
    return Flowable.error(error);
}

But biggest problem there is blockingGet(). I believe this is very bad solution. Maybe anyone know any better one? 

