使用 RX Java 运算符设计 API 流程

Question

我是 RX-Java 的新手，我正在尝试设计一个 API，其流程如下所述：

            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

假设 DB 方法和服务调用返回 Observable，需要对上面使用 rx 运算符的框架流程进行一些澄清吗？

我将在下面分享阻塞伪代码：

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);
}

我正在研究一种完整的非阻塞异步方法。

Answer 1

一般的逻辑流程可以如下：

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))
 )

请注意，如果流程中有复杂的逻辑，我总是会尝试将其提取到单独的方法中。在您的情况下，基于 REST 调用调用数据库可能涉及一些转换 - 如果生成的逻辑超过一两行，我会将其移至单独的方法并在 RX 流中使用方法引用。

我们的想法是让流程成为可以在单个页面中查看和解析的东西，并在方法中隐藏实现细节。

编辑：编辑后，也许这更有意义：

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));
 }

Answer 2

Tassos 的回答在我看来是合理的，我没有看到提到的任何其他问题，但是

// dataSource.getItemDetails(activityId returns List<ItemDetail> and is a blocking call
// So, I want to run it on a separate IO thread.
return Observable.from(dataSource.getItemDetails(activityId)).observeOn(Schedulers.io());

如果是这样，将阻塞的单元素调用转换为线程外调用可以如下完成：

Observable.fromCallable(() -> yourBlockingCall())
.subscribeOn(Schedulers.io())
.flatMapIterable(v -> v)
...

或

Observable.defer(() -> Observable.from(yourBlockingCall()))
.subscribeOn(Schedulers.io())
...

编辑：根据图表，我将设置以下流程：

serviceCallA()
.flatMap(a -> {
    if (dataFound(a)) {
        return dbCall1()
           .flatMap(db1 -> {
               Observable o1 = shouldCallServiceC(db1) 
                    ? serviceCallC() : just(placeholderC);
               Observable o2 = shouldCallDB2(db1) 
                    ? dbCall2() ? just(placeHolderDb2);

               return zip(o1, o2, (c, d) -> createResult(c, d));   
           });
    }
    return serviceCallB()
        .map(c -> mapToResultType(c));
});