RxJava 从 Observable<JsonArray> 返回 JsonArray

Question

我对函数式编程和反应式 RxJava 还很陌生。我想从数据库中获取设备的 ID 和名称并将其存储在 Map 中，我正在以 RxJava 风格进行操作。我正在调用一个不需要返回任何内容的函数

.doOnNext(t -> updateAssetNameMap())

然后函数看起来像;

    private void updateDeviceNameMap() {
    LOGGER.debug("Reading device name and id from database");
    Observable<SQLConnection> jdbcConnection = createJdbcConnection();
    Scheduler defaultScheduler = RxHelper.scheduler(vertx);
    Observable<JsonArray> res =  jdbcConnection  //need to return JsonArray
            .flatMap(connection -> just(connection)
                    .flatMap(j -> runQuery(connection, "SELECT name,id FROM device")
                                .observeOn(defaultScheduler)
                                .doOnNext(m -> LOGGER.info("size: " + m.size()))
                                .flatMap(job -> { LOGGER.info(">>" + job.getJsonArray(0));
                                         //or if I can extract JsonArray items here, 
                                         //I can update my Map here too.
                                                  return just(job.getJsonArray(0));
                                                }
                                       )
            .doOnError(e -> { LOGGER.error("failed to connect to db", e);
                        connection.close(); })
            .doOnCompleted(connection::close)
            .onErrorReturn(e -> null));

    //System.out.println("" + res.map(d -> LOGGER.info(d.toString())));
    //get the JsonArray and update the deviceNameMap

与数据库的连接成功，查询也正确完成。 我可以通过Observable.from(ObjectName) 将任何对象转换为 Observable，但不能相反。在.flatMap(job -> just(job.getJsonArray(0)) 之后需要进行适当的映射，但我不知道如何。运行 Verticle 后，我什至看不到从行 .flatMap(job -> { LOGGER.info(">>" + job.getJsonArray(0)); 记录的任何内容。 我错过了什么吗？

Answer 1

您必须订阅您的Observable<JsonArray>，否则不会发生任何事情。