Design API flow using RX Java Operators

Question

I'm new to RX-Java and I'm trying to design an API where the flow is as mentioned below :

            Make REST call A to load data
                        |
                        |
        data not found? |  data found  
    ------------------------------------    
    |                                  |           
    |                                  |            
    |                                  |
Make REST Call B                  Load DB Data 1     
    |                                  |            
    |                                  |
    |                              _________________________
    |                             |       Parallel         |
    |                             |                        |
    |                             |                        |
    |                  (condition using DB data 1)   (condition using DB data 1)
    |                      Load REST Data C                Load DB Data 2
    |                             |                        |
    |                             |________________________|
    |                                         |
    |                                         |
Build Response                            Build Response

Assuming DB methods and service calls returns Observable, need some clarity with the skeleton flow for above using rx Operators ?

I'll share the blocking pseudo code below :

Response = REST_Call_1(); // on error throw Exception

if (isResponseValid(response)) { // returns Boolean
    if (responseUnderReview(response)) { // validation func 
        throw Exception;
    } else {
        //db_data_1 and db_data_2 can be parallel
        db_data_1 = Load_DB_Data_1();

        // Load data_3 and data_2 based on db_data_1
        if (is_data_3_required(db_data_1)) {
            data_3 = REST_call_2();
        }
        if (is_data_2_required(db_data_1)) {
            db_data_2 = REST_call_2();
        }
        buildResponse(db_data_1, db_data_2, data_3, Response);
    }
} else {
    Response = REST_Call_3(); // on error throw Exception
    buildResponse(response);
}

I'm looking at a complete non-blocking asynchronous approach.

Answer 1

The general flow of the logic can be as follows:

retrofitClient
.loadData()...
.onErrorResumeNext(Observable.empty()) // or handle specific errors only
.flatMap(foundData -> 
    Observable.zip(
       database.call1(foundData),
       database.call2(foundData),
       (call1, call2) -> buildResponse(call1,call2)
    )
 )
 .switchIfEmpty(() ->
    retrofitClient
    .callB()
    .map(response -> buildResponse(response))
 )

Note that if there's complex logic in the flow I always try to extract it into separate methods. In your case, calling the database based on the REST calls might involve some conversions - if the resulting logic is more than a line or two, I'd move it to a separate method and use a method reference in the RX flow.

The idea is to have the flow as something that can be looked at and parsed in a single page, and hide implementation details in methods.

Edit: After your edit, maybe this makes more sense:

REST_Call_1()
.filter(response -> isResponseValid(response))
.flatMap(response ->
     isResponseUnderReview(response)
     ? Observable.error(new Exception())
     : Observable.just(response)
)
.flatMap(foundData -> 
    Observable.zip(
       fetchData13(foundData),
       Load_DB_Data_2(foundData),
       (data13, call2) -> buildResponse(data13.getLeft(),call2,data13.getRight())
    )
 )
 .switchIfEmpty(() ->
    REST_Call_3()
    .flatMap(response -> buildResponse(response))
 )
 .subscribe(....)

 private Observable<Pair<DbData1, DbData3>> fetchData13(foundData) {
    return
    Load_DB_Data_1()
    .flatMap(data1 -> is_data_3_required(data1)
        ? REST_call_2().map(data3 -> Pair.of(data1, data3))
        : Pair.of(data1, null));
 }