Why Webflux 'then' operator not execute?

public Mono<Integer> saveAccount(Mono<AccountSaveRequest> requestMono) {
    
    var reqMono = requestMono.map(AccountSaveRequest::convertEntity);
    
    return reqMono
       .flatMap(req -> this.accountRepository.findByAccount(req.account())
                 .doOnNext(model -> {
                     if (req.id() == null || !model.id().equals(req.id())) {
                         throw new BusinessException("account exists");
                     }
                  })
        )
       .then(reqMono)
       .flatMap(req -> this.accountRepository.findByMobile(req.mobile())
                 .doOnNext(model -> {
                     if (req.id() == null || !model.id().equals(req.id())) {
                         throw new BusinessException("mobile exists");
                     }
                 })
        )
       .then(Mono.defer(() -> reqMono.flatMap(this.accountRepository::save)))
       .map(AccountEntity::id);
}

See code, easy to understand, this method is used to save a user to database.

First receive a AccountSaveRequest model from request and mapped to AccountEntity, then check if
'account' field is already exists in database.

If not exists, I use 'then(reqMono)' operator to switch to another Mono, the following Mono is used to check the field 'mobile' is already exists in database if mobile not blank.

If mobile still not exists, I finally save the account to database and return id to consumer.

The method will return an id if everything is ok. Howerver the code after 'then(reqMono)' not execute.

I tried this in another way，see code

public Mono<Integer> saveAccount(Mono<AccountSaveRequest> requestMono) {
    return requestMono.map(AccountSaveRequest::convertEntity)
       .flatMap(req -> this.accountRepository.findByAccount(req.account())
                  .doOnNext(model -> {
                      if (req.id() == null || !model.id().equals(req.id())) {
                          throw new BusinessException("account exists");
                      }
                  })
                  .switchIfEmpty(
                       Mono.defer(() -> this.accountRepository.findByMobile(req.mobile())
                           .doOnNext(model -> {
                                if (req.id() == null || !model.id().equals(req.id())) {
                                    throw new BusinessException("mobile exists");
                                }
                           })
                       )
                  )
                  .then(Mono.defer(() -> this.accountRepository.save(req)))
                  .map(AccountEntity::id)
        );
}

I wrapped the the code after 'then(reqMono)' into the first flatMap call, and it works correctly

Can find the problem? THKS

In the first case it doesn't work because you're trying to consume reqMono twice. I.e., you subscribe on it (flatMap) twice. Which doesn't work because for the second subscription it's already produced the result and complete.

You can do reqMono.cache() before the subscriptions and then the first way would work. Thought that is not very reactive way.

The second example is how it supposed be with Reactor. You process an item once and then produce results based on that data.

PS. The other things that don't look good are:

1. You use doOnNext for side effects. That's also is not how it supposed to work. Use flatMap if you want to produce different outcome based on the input.
2. You throw exceptions. Use Mono.error instead.
3. Your Mono.defer make no sense. Just call findByAccount, findByMobile and save directly, since they already give a Producer

So I think your code should look like this:

public Mono<Integer> saveAccount(Mono<AccountSaveRequest> requestMono) {
    return requestMono
       .map(AccountSaveRequest::convertEntity)
       .flatMap(req -> 
           this.accountRepository.findByAccount(req.account())
              .switchIfEmpty(this.accountRepository.findByMobile(req.mobile()))
              .flatMap(model ->
                 if (req.id() == null || !model.id().equals(req.id())) {
                     Mono.error(new BusinessException("exists"))
                 } else {
                     Mono.empty()
                 }
              )
              .switchIfEmpty(this.accountRepository.save(req))
              .map(AccountEntity::id)
       )            
 }