为 Object Flux 实现 Upsert 功能答案

【问题标题】：Implementing Upsert functionality for Object Flux为 Object Flux 实现 Upsert 功能
【发布时间】：2020-04-22 10:11:36
【问题描述】：

我有一个包含对象的通量请求。任务是创建（如果不存在）或更新（如果存在）这些对象。我已经尝试如下实现：

requestFlux.flatMap(entry -> {
        Mono<Optional<MyObject>> existingMapping = myRepository
            .findByIdAndName(entry.getId(), entry.getName());
        return existingMapping.map(optional -> {
        if (optional.isPresent()) {

            MyObject model = MyObjectMapper.updateMyObject(entry, optional.get()); 
            myRepository.save(model);
            return model;
        } else {

            MyObject model = MyObjectMapper.toMyObjectModel(entry);
            myRepository.save(model);
            return model;
        }
        });

    });

由于某种原因，这似乎不起作用。它不会引发任何错误，不会创建/更新记录或记录 if/else 中存在的消息。想知道我错过了什么吗？

【问题讨论】：

myRepository.save(model); 返回什么？我的意思是什么是返回类型？你会返回requestFlux.flatMap(entry -> {....}); 这整个链条吗？
save() 方法返回 MyObject 类型的 Mono。我没有返回您提到的整个链，因为该方法的返回类型与该链返回的类型不同。
你想从整个链条中返回什么？我认为有一个简单的解决方案
嗨@Shoshi，RequestFlux 包含Objects，需要通过调用不同的转换函数将其转换为MyObjects。然后这些 MyObjects 需要通过save() 函数保存。我也在尝试找出简单的方法，但还没有弄清楚。
最后你想要返回的究竟是什么？

标签： rx-java spring-webflux project-reactor nonblocking

【解决方案1】：

您必须在一次更改中绑定所有内容。不要破坏链条。

public Mono<ServerResponse> createOrUpdateBulkMapping(ServerRequest serverRequest) {

    Flux<RequestObjects> requestFlux = serverRequest.bodyToFlux(RequestObjects.class);
    List<MyObjects> createdList = new ArrayList<>();
    List<MyObjects> updatedList = new ArrayList<>();

    return requestFlux
            .flatMap(entry -> {

                return myRepository.findByIdAndName(entry.getId(),entry.getName())
                       .flatMap(optional -> {
                          updatedList.add(entry);
                          MyObject model = MyObjectMapper.updateMyObject(entry, optional.get());
                          return myRepository.save(model);
                       })
                       .switchIfEmpty(saveNewModel(entry, createdList));

            })
            .then(ServerResponse.ok().body(Mono.just("Bulk Create/Update is successful").log(), String.class));
}

private Mono<MyObject> saveNewModel (RequestObjects entry, List<MyObjects> createdList) {
  createdList.add(entry);
  MyObject model = MyObjectMapper.toMyObjectModel(entry);
  return myRepository.save(model);
}

已编辑：作为您在此处提供的要点：https://gist.github.com/vkrmsngh43/13a8753848893a0f6988d8327f656220 答案已更改

【讨论】：