SpringWebflux. Switch between reactive streams

Question

I try to create a new record in DB but before I need to fetch another record and take some data from that and set it to a new record. It supposes Mono<Subject> subject will fill data in subject.doOnNext(child -> {...}) and after that inserted to DB with subjectRepository::insert, but nothing happens and no error, the object is just not inserted.

public Mono<Subject> create(Mono<Subject> subject) {
    return subject
            .flatMap(s -> subjectRepository.findById(s.getParentId()))
            .flatMap(parent ->
                    subject.doOnNext(child -> {
                        child.setParentId(parent.getId());
                        child.setCreatedDate(LocalDateTime.now());
                        child.setRoute(Utils.buildRoute(child, parent));
                    }))
            .flatMap(subjectRepository::insert);
}

What is my mistake and how to fix it?

Answer 1

subject.doOnNext is never subscribed, thus is never actually executed. Also, doOn*** operators are "side effects" operators and you should not implement business logic or I/O operations within those. You could try something like this:

return subject
    .flatMap(s -> subjectRepository.findById(s.getParentId())
        .map(parent -> {
          s.setParentId(parent.getId());
          s.setCreatedDate(LocalDateTime.now());
          s.setRoute(Utils.buildRoute(child, parent));
          return s;
        }))
    .flatMap(subjectRepository::insert);

Answer 2

Try this snippet.

return subject
    .flatMap(s -> {
        return repository.findById(s.getParentId())
                .flatMap(p-> {
            s.setParentId(p.getId());
            s.setCreatedDate(LocalDateTime.now());
            s.setRoute(Utils.buildRoute(s, p));
            return repository.save(s);
        });
    });