java - java rxObservable 的结果集中没有聚合数据

Question

我尝试调用存储过程 3 次。当我运行下面的代码时，上次调用存储过程的数据仅显示在 resultSet.getRows() 中。前两次调用存储过程的数据没有出现在结果集中。以下是我的代码。难道我做错了什么。谁能帮忙？

String currentPeriod = String.format("{call %s.testProc(?)}", params.getJsonObject("databaseInfo").getString("dbName"));
String priorPeriod   = String.format("{call %s.testProc(?)}", params.getJsonObject("databaseInfo").getString("dbName"));
String todayPeriod   = String.format("{call %s.testProc(?)}", params.getJsonObject("databaseInfo").getString("dbName"));
 JsonArray jsonArray = new JsonArray();

        database.dbObject().getConnectionObservable().subscribe(
                connection -> {
                    Observable<ResultSet> resultSetObservable = connection.callWithParamsObservable(currentPeriod, new JsonArray().add(params.getString("testParams")),jsonArray ).                                                                              
                            flatMap(result -> connection.callWithParamsObservable(priorPeriod, new JsonArray().add(params.getString("testParams")), jsonArray ).
                            flatMap(result -> connection.callWithParamsObservable(todayPeriod, new JsonArray().add(params.getString("testParams")),jsonArray );

                    resultSetObservable.subscribe(resultSet -> {
                        handler.handle(ReportUtils.parseSQLResult(resultSet.getRows()));
                    },error -> {
                        error.printStackTrace();
                    },connection::close);

                },err -> {
                    err.printStackTrace();
                }
        );

Answer 1

您尝试做的事情可以通过使用 combineLatest 运算符来实现，该运算符给出所有可观察对象的结果(它将等待所有可观察对象给出结果)

引用 http://reactivex.io/documentation/operators/combinelatest.html

伪代码，

       String currentPeriod = String.format("{call %s.testProc(?)}", params.getJsonObject("databaseInfo").getString("dbName"));
      String priorPeriod   = String.format("{call %s.testProc(?)}", params.getJsonObject("databaseInfo").getString("dbName"));
      String todayPeriod   = String.format("{call %s.testProc(?)}", params.getJsonObject("databaseInfo").getString("dbName"));
     JsonArray jsonArray = new JsonArray();


    database.dbObject().getConnectionObservable().subscribe(
            connection -> {
                resultSetObservable = Observable.combineLatest(firstCall, secCall, thirdCall)
     firstCall = connection.callWithParamsObservable(currentPeriod, new JsonArray().add(params.getString("testParams")),jsonArray )

      secCall = result -> connection.callWithParamsObservable(priorPeriod, new JsonArray().add(params.getString("testParams")), jsonArray )

     thirdCall = result -> connection.callWithParamsObservable(todayPeriod, new JsonArray().add(params.getString("testParams")),jsonArray )
                resultSetObservable.subscribe(firstRes, secRes, thirdRes -> {
                    handler.handle(ReportUtils.parseSQLResult(resultSet.getRows()));
                },error -> {
                    error.printStackTrace();
                },connection::close);

            },err -> {
                err.printStackTrace();
            }
    );