Java Reactor - conditional stream execution

Question

I was wondering how to create "logical stream" using Reactor.

Lets assume that I want to implement following scenario:

As input I have object to save in database. As output I would like to get Mono representing execution message.

• Option 1: if object to save has all fields filled then I perform additionall operations, save it to database and finally return "Success" message

• Option 2: if object to save has at least one field not filled I return "Error"

I have created such code:

Mono<String> message = Mono.just(new User("Bob the Reactor master")) // user with name = can be saved
    .flatMap(user -> {
        if(user.getName() != null && user.getName().length() > 1){
            // Perform additional operations e.g. user.setCreatedDate(new Date())
            // Save to repository e.g. repository.save(user)
            return Mono.just("Success!");
        }
        else{
            return Mono.just("Error!");
        }
    })
    .doOnNext(System.out::println); // print stream result

message.subscribe();

Is this code 100% reactive (has all its benefits)? If no then what it will look like?

Answer 1

The answer depends on your commented repository.

• Repository is non-blocking and returns Mono or Flux

  You should subscribe it then return Success Mono. In your if statement:

  return repository.save(user).then(Mono.just("Success!"));
  

• Repository is blocking

  You should make your repository call non-blocking moving its execution to separate thread. Reactor way is to wrap it with Mono and subscribe on elastic scheduler or your custom scheduler.

  Mono<String> message = Mono.just(new User("Bob the Reactor master"))
          .filter(user -> user.getName() != null && user.getName().length() > 1)
          .map(user -> user) // Perform additional operations e.g. user.setCreatedDate(new Date())
          .flatMap(user -> Mono.fromRunnable(() -> repository.save(user))
                  .subscribeOn(Schedulers.elastic())
                  .then(Mono.just("Success!")))
          .switchIfEmpty(Mono.just("Error!"));