RXJava Handle chain api calls: show different error message and continue the stream

Question

I'm trying to convert a callback hell to RX but I'm stuck with getting the proper order, below is my functionality I want to achieve

a) User Login-> get the Auth Cookies, if login credentials invalid show error message

b) use the Auth Cookies to get Customer Type,

c) if the Customer Type is zero/ show profile Restricted Error Message and log out the user

d) if the customerType, not zero proceed to get the other customer Details

e) if any of the customer APIs returns an error response, log out the user and show login failure message

f) if all customer API success show the home screen

API 
Login

      @FormUrlEncoded
        @POST("distauth/UI/Login")
        Single<Response<Void>> doLogin1(@Field("username") String username, @Field("password") String password,
                                       @Field("rememberme") String rememberMe, @Field("answer") String answer,
                                       @QueryMap Map<String, String> options);

    public Single<Boolean> doLogin(@NonNull String username, @Nullable String password) {
        return authapi.doLogin1(username, password, "y", "", logiOptions)
                .flatMap(new Function<Response<Void>, SingleSource<Boolean>>() {
                    @Override
                    public SingleSource<Boolean> apply(Response<Void> response) throws Exception {
                        if (response.code() == HttpStatus.MOVED_TEMPORARILY.value()
                                && !StringUtils.isEmpty(Session.getCookie())
                     ) {
                            return Single.just(true);
                        }
                        throw new Exception("Invalid Login Details");
                    }
                });
    }

     //==========
    Logout
        @FormUrlEncoded
        @POST("distauth/UI/Logout")
        @Headers("Cache-Control: no-cache")
        Completable doLogout(@Field("logout") boolean logout); //return 302 HTTP Status code with empty iPlanetCookie
     //==========
    NOTE: Loing/logout is not a REST API, this legacy app implement as Form Post ;) so when the success of login return 302 with cookies, and log out also return 302 as status code 

    Get Customer Details

      Single<CustomerAccountVO> getCustomerAccountDetails(boolean forceRefresh);
     //==========
       Single<CustomerType> getCustomerUserProfile(boolean forceRefresh);

    @Override
    public Single<CustomerType> getCustomerUserProfile(boolean applyResponseCache) {
        return this.mCustomerRemoteDataStore.getCustomerUserProfile(applyResponseCache)
                .doOnSuccess(new Consumer<CustomerType>() {
                    @Override
                    public void accept(CustomerType customerType) throws Exception {
                        if (customerType != null && customerType.getBody() != null &&
                                !StringUtils.isEmpty(customerType.getBody())) {
                            if (customerType.getBody().equalsIgnoreCase(AppConfig.ERROR)) {
                                throw new CustomerProfileNotFound(500, "user account restrictions");
                            } else {
                                mCustomerLocalRepository.saveCustomerType(customerType);
                            }
                        }
                    }
                }).doOnError(new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.e(TAG, "error occurred while getting customer user profile", throwable);
                    }
                });
    }

     //==========
        Single<CustomerAccountId> getAccountId(boolean forceRefresh);
     //==========
        Single<Customer> getCustomer(boolean forceRefresh);
     //==========
    Get Customer Full Details
        Single<CustomerDetails> getCustomerFullDetails(boolean applyResponseCache);

    Implementation: 

        @Override
        public Single<CustomerDetails> getCustomerFullDetails(boolean forceRefresh) {
            Single<CustomerDetails> customerDetails = Single.zip(
                    getCustomerUserProfile(forceRefresh).subscribeOn(Schedulers.io()),
                    getAccountId(forceRefresh).subscribeOn(Schedulers.io()),
                    getCustomerAccountDetails(false).subscribeOn(Schedulers.io()),
                    getCustomer(forceRefresh).subscribeOn(Schedulers.io()), new Function4<CustomerType, CustomerAccountId,
                            CustomerAccountVO, Customer, CustomerDetails>() {
                        @Override
                        public CustomerDetails apply(@NonNull CustomerType customerType,
                                                     @NonNull CustomerAccountId customerAccountId,
                                                     @NonNull CustomerAccountVO customerAccountVO,
                                                     @NonNull Customer customer) throws Exception {
                            return CustomerDetails.builder().customerType(customerType).customerAccountVO
                                    (customerAccountVO).customer(customer).customerAccountId(customerAccountId).
                                    build();
                        }
                    });

            return customerDetails;
        }
     //==========
    Each customer request is independent so I thought to execute as sperate thread and zip the final result/

        Single<BaseServerResponse> updateCustomerDetails(@Nonnull boolean secure, int secureRequestCode, @Nonnull JSONObject customerContact);


    //Presenter Implementation: this implementation not working as i expect above, can some one help me to get this correct,  
     public void doLoginHandler(@NonNull String username, @NonNull String password) {
    checkViewAttached();
    getMvpView().showLoadingIndicator();
    addSubscription(
            apiService.doLogin2(username, password)
                    .subscribeOn(Schedulers.io())
                    .flatMap(new Function<Boolean, SingleSource<CustomerDetails>>() {
                        @Override
                        public SingleSource<CustomerDetails> apply(Boolean aBoolean) throws Exception {
                            if (aBoolean) {
                                //get customr Full Details
                                Log.d(TAG, "apply: "+aBoolean);
                                return customerRepository.getCustomerFullDetails(true);

                            }
                            return null;
                        }
                    }).observeOn(AndroidSchedulers.mainThread())
                    .onErrorResumeNext(new Function<Throwable, SingleSource<? extends CustomerDetails>>() {
                        @Override
                        public SingleSource<? extends CustomerDetails> apply(Throwable throwable) throws Exception {
                            if (throwable instanceof CustomerProfileNotFound) {
                                getMvpView().showUserProfileAccessRestrictMessage();
                            } else {
                                getMvpView().onLoginAuthFailure();
                            }
                            return Single.just(CustomerDetails.builder().errorOccurred(true).build());
                        }
                    })
                    .flatMapCompletable(new Function<CustomerDetails, CompletableSource>() {
                        @Override
                        public CompletableSource apply(CustomerDetails customerDetails) throws Exception {
                            if(customerDetails.isErrorOccurred()){
                                return apiService.doLogout();
                            }
                            return Completable.complete();
                        }
                    })
                    .subscribe(new Action() {
                        @Override
                        public void run() throws Exception {
                            getMvpView().onLoginAuthSuccess();
                        }
                    }, new Consumer<Throwable>() {
                        @Override
                        public void accept(Throwable throwable) throws Exception {
                            if (throwable instanceof CustomerProfileNotFound) {
                                getMvpView().showUserProfileAccessRestrictMessage();
                            } else {
                                getMvpView().onLoginAuthFailure();
                            }
                        }
                    }));


}

Ahmed Ashraf Ahmed Ashraf · Accepted Answer · 2018-07-22T03:33:22

First I'll state the problem with your code.

.flatMapCompletable(new Function<CustomerDetails, CompletableSource>() {
                    @Override
                    public CompletableSource apply(CustomerDetails customerDetails) throws Exception {
                        if(customerDetails.isErrorOccurred()){
                            return apiService.doLogout();
                        }
                        return Completable.complete();
                    }
                })

This chain observable (which is the one you subscribe to) is always going to give a Completed state unless a network error happens when calling the logout API, that's because you either return the logout Completable or an instant Completable.

Secondly, I think the solution is in logically sorting everything out, the key to error handling in such a case would be creating a different Exception for each error case with it's own error message,

it can go like this (I'm just using the logical names, hopefully that will give you the idea):

loginObservable.flatMap { authCredentials -> { 
if (authCredentials.isValid())
    return getCustomerTypeObservable(authCredentials)
else
    return Single.error(InvalidCredentialsException("message goes here (optional)"))
}}.flatMap { type -> {
    if (type == 0)
        return Single.error(ProfileRestrictedException("different message maybe?"))
    else
        return getCustomerDetailsZippedObservable(type)
}}
/* ..etc */

Then at the subscription site you do something like:

myObservable.subscribe( {
/* Handle success*/
}, { exception ->
    when(exception) {
        is InvalidCredentialsException -> mvpView.showError(message)
        is ProfileRestrictedException -> { 
            mvpView.showError(message)
            logout()
        }
        else -> /* Handle an exception that is not listed above */
    }

} )

This way IMO is more convenient than using onErrorResumeNext.

EDIT: You can also overcome the issue stated above by doing something like:

.flatMapCompletable { customerDetails -> {
    if(customerDetails.isErrorOccurred()){
        return apiService.doLogout()
                .then(Completable.error(LoginFailedException("Message"))) /* This will guarantee the stream terminates with the required error type after logout is successful */
    } else {
        return Completable.complete()
    }
}}

RXJava Handle chain api calls: show different error message and continue the stream

1 Answers