如何使用 Reactor 保存嵌套实体？

如何解决如何使用 Reactor 保存嵌套实体？

我将 Spring 与 R2DBC 一起使用，我需要执行以下操作：

for (var q: quiz.quizzes()) {

  Set<UUID> qsIds = new LinkedHashSet<>();

  for (var qs: q.questions()) {
   var ansids = answerRepository
            .saveAll(mapToAnswerEntity(qs.getAnswers()))
            .collectList().toFuture()
            .get()
            .stream()
            .map(AnswerEntity::id)
            .collect(Collectors.toSet());
   var qe = QuestionEntity.builder().id(qs.getId()).answerIds(ansids).title(qs.getTitle()).build();
   var id = questionRepository.save(qe).toFuture().get().id();
   qsIds.add(id);
  }
  var newQ = quizRepository
          .save(
                  toQuizEntity(q,qsIds)).toFuture().get();

 var updatedCourse =  courseRepository.findById(courseId).toFuture().get().addQuizToCourse(newQ.id());
  courseRepository.save(updatedCourse);
}

return Mono.empty();

问题是我不能使用 .get() 因为 Reactor 要求（我使用的是 Webflux）。

我尝试了以下方法：

return Flux.fromIterable(quiz.quizzes())
        .flatMap(q-> Flux.zip(Flux.just(q),Flux.fromIterable(q.questions())))
        .flatMap(t->
                Flux.zip(Flux.just(t.getT1()),Flux.defer(()->Flux.just(t.getT2())),Mono.from(answerRepository.saveAll(mapToAnswerEntity(t.getT2().getAnswers())).collectList())))
        .flatMap(t->
                Flux.zip(
                        Flux.just(t.getT1()),Flux.just(toQuestionEntity(t)).collect(Collectors.toSet())))

        .flatMap(t-> Flux.zip(Flux.just(t.getT1()),questionRepository.saveAll(t.getT2()).collectList()))
        .map(t-> toQuizEntity(t.getT1(),t.getT2().stream().map(QuestionEntity::id).collect(Collectors.toSet())))
        .flatMap(t-> quizRepository.save(t))
        .flatMap(q-> courseRepository.findById(courseId).map(x->x.addQuizToCourse(q.id())))
        .flatMap(c-> courseRepository.save(c))
        .then(Mono.empty());

但是，这并没有返回我期望的所有结果。基本上只保存相对于第一个测验的第一个或最后一个结果（问题和答案）。 2 个测验（我用于测试目的）保存在数据库中，只是它们的“依赖项”受到影响。 2 个问题（每个测验 1 个）保存了其他问题。答案几乎相同。

编辑

我试过这个方法，解决了之前的问题，但是现在quiz_ids列只添加了一门课程。

return Flux.fromIterable(quiz.quizzes())

        .flatMap(q-> Flux.fromIterable(q.questions())
                        .flatMap(qs-> answerRepository
                                .saveAll(mapToAnswerEntity(qs.getAnswers()))
                                        .collect(Collectors.toSet())
                                        .flatMap(l-> Mono.just(l.stream().map(AnswerEntity::id).collect(Collectors.toSet())))
                                        .flatMap(l -> Mono.defer(()-> Mono.just(toQuestionEntity(qs,l)))))
                                .collect(Collectors.toSet())
                                        .flatMap(l-> questionRepository.saveAll(l).collect(Collectors.toSet()))
                                        .map(l->l.stream().map(QuestionEntity::id).collect(Collectors.toSet()))
                                        .flatMapMany(qsl-> quizRepository.save(toQuizEntity(q,qsl)))
                                        .doOnNext(System.out::println))
        .flatMap(qe->Flux.defer(()->courseRepository.findById(courseId).map(x->x.addQuizToCourse(qe.id())).flatMap(c->courseRepository.save(c))))
        .doOnNext(System.out::println)
        .then(Mono.empty());

这是在课程中添加测验的原因： x.addQuizToCourse(qe.id())

这是日志记录的结​​果：

    QuizEntity(id=e9e13855-2c42-456e-9fab-a0865ad96a3f,quizName=1111,quizContent=julia2020 2 test quiz content,questionIds=[a5d576f0-afc9-4312-9f29-cdc018eb635a,cc7282c5-ffd0-4750-9180-ca5e50d4f515])
QuizEntity(id=73315bac-dce0-4595-91c2-85fe2a2e4765,quizName=2222,quizContent=julia2020 test quiz content,questionIds=[b4b1765e-e41a-427b-9073-ad2b16dabe3a,0e9ad984-b63a-4ca4-aa12-efa1a3e7ae6a])
CourseAggregate(id=776fff36-34d9-4b53-8395-c72ee4816f3f,courseName=Mammia course,description=<p>Test Courseasdsadsadsasadasdsa</p>,status=DRAFT,teacherId=79ab22d9-af94-49f1-a25e-c74bada7ce6f,studentIds=[],lessonIds=[4f364a50-bc28-4d75-9f76-71ff43d30ebf,eaaab4ae-6ab4-4a14-a906-259293780e17,28db0295-82ab-4074-aff7-5e10374191b5],quizIds=[e9e13855-2c42-456e-9fab-a0865ad96a3f])
CourseAggregate(id=776fff36-34d9-4b53-8395-c72ee4816f3f,quizIds=[73315bac-dce0-4595-91c2-85fe2a2e4765])

如您所见，最后一门课程汇总不包含前一课程，而仅包含当前测验 ID。

编辑 2： 我用 delayUntil 解决了这个问题：

return Flux.fromIterable(quiz.quizzes())

        .flatMap(q-> Flux.fromIterable(q.questions())
                        .flatMap(qs-> answerRepository
                                .saveAll(mapToAnswerEntity(qs.getAnswers()))
                                        .collect(Collectors.toSet())
                                        .flatMap(l-> Mono.just(l.stream().map(AnswerEntity::id).collect(Collectors.toSet())))
                                        .flatMap(l -> Mono.defer(()-> Mono.just(toQuestionEntity(qs,qsl)))
                                        .doOnNext(System.out::println))
        .delayUntil(qe-> courseRepository.findById(courseId).map(x->x.addQuizToCourse(qe.id())).flatMap(c->courseRepository.save(c)))
        .doOnNext(System.out::println)
        .then(Mono.empty());