如何从其他类型的地图返回对象？答案

【问题标题】：How to return object from map of some other type?如何从其他类型的地图返回对象？
【发布时间】：2019-05-19 15:55:45
【问题描述】：

我正在尝试编写批量保存或更新类型的功能。从 CSV 文件中读取记录并转换为适当的类，然后为该记录执行保存或更新。

以下是保存或更新的步骤

通过 emailId 从 DB 获取。（可能返回）
如果用户存在，则调用 save else 进行更新。
如果 DB 操作成功，则返回 Report 类型的 Single。这里的 Report 是一个用户定义的类。

 Flowable.fromPublisher(userRepository.findByEmail(user.emailId)).firstElement()

此代码返回 MayBe，对此，我想调用 Save/Update，它返回 Single 或错误。无论如何，我想返回类型为 Report 的 Single

以下是我尝试过阻塞方式的代码示例，但我想以非阻塞方式进行

   @Override
    Single<BulkUploadReport> processData(String nextLine) {
        List<Object> errors = new ArrayList<>()

        Single.create({ emitter ->

            def parsedString = parseCsv(nextLine, readFirstLine: true, columnNames: columnNames)
            for (record in parsedString) {


                try {
                    User user = parseRecord(record)
                    List<String> roles = parseRoles(record[14])


                   User dbUser =  Flowable.fromPublisher(userRepository.findByEmail(user.emailId)).firstElement().blockingGet()

                    if(dbUser == null){
                         userService.save(user, user.emailId, roles, saveUserkey).blockingGet()

                    }else{
                        user.userkey = dbUser.userkey
                         Single.fromPublisher(userRepository.updateUser(saveUserkey, user)).blockingGet()

                    }
                    emitter.onSuccess(new BulkUploadReport(record[0] as int, true, null))
                } catch (Exception e) {
                    log.error("Error in bulk update.Handled grac8efully:  ${e.getMessage()}")
                    ArrayList<String> err = e.getMessage().tokenize(",") as ArrayList<String>
                    errors.addAll(err)
                    emitter.onSuccess(new BulkUploadReport(record[0] as int, false,errors))
                }

            }

        }) as Single<BulkUploadReport>
    }

用户存储库：

 Publisher<User> updateUser(String saveUserKey, User user)
 Publisher<User> findByEmail(String emailId)
 Publisher<Success> insertUser(User user)

我是 RxJava 新手，我不明白如何链接这些事件。

【问题讨论】：

你能贴出userRepositiry函数的签名，用在那个sn-p吗？顺便说一句，当你使用Reactive Streams 中的Single 类型时，你应该尊重它的联系方式——onSuccess 应该只被调用一次或从不调用。但是你在 for 循环中调用它。在这种情况下，最好使用Observable。
@ConstOrVar 更新了问题

标签： rx-java2

【解决方案1】：

在我看来，你应该避免所有的blockingGet() 调用（这不是 Rx 的真正理念）并尝试使用 Rx 方法调用链接进程。在您的代码 sn-p 中，不清楚您是在尝试处理来自 CSV 的用户列表还是单个用户。可能如果你能给parseCsv()的签名，那就更清楚了。也不清楚哪个方法可以抛出异常。

如果曾经从 CSV 中提取单个用户，这里有一个伪代码答案可能会起到作用：

public Single<BulkUploadReport> processData(String line) {
    final Record record = parseCsv(line, true, columNames);
    final User user = parseRecord(record);
    final List<String> roles = parseRoles(record);

    return Flowable.fromPublisher(userRepository.findByEmail(user.emailId))
                   .firstElement()
                   .switchIfEmpty(Maybe.just(userService.save(user, user.emailId, roles, saveUserkey)))
                   .map(aUser -> new BulkUploadReport(record, false))
                   .toSingle();
}

希望这会有所帮助。

【讨论】：