Java Reactor - условное выполнение потока

Question

Мне было интересно, как создать «логический поток» с использованием Reactor.

Допустим, я хочу реализовать следующий сценарий:

В качестве ввода у меня есть объект для сохранения в базе данных.В качестве вывода я хотел бы получить Mono, представляющее сообщение выполнения.

• Опция 1: если для объекта сохранения заполнены все поля, тогда я выполняю все дополнительные операции, сохраняю их вбазы данных и, наконец, вернуть сообщение "Success"

• Опция 2: если для объекта сохранения хотя бы одно поле не заполнено, я возвращаю "Error"

Я создал такой код:

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();

Является ли этот код на 100% реактивным (имеет все свои преимущества)?Если нет, то как это будет выглядеть?

Answer 1

Ответ зависит от вашего закомментированного репозитория.

• Репозиторий неблокируемый и возвращает Mono или Flux

  Вы должны подписаться, а затем вернуть Success Mono.В вашем операторе if:

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

• Хранилище блокирует

  Вы должны сделать неблокирующим вызов репозитория, перенеся его выполнение в отдельный поток.Способ Reactor заключается в том, чтобы обернуть его с помощью Mono и подписаться на эластичный планировщик или ваш собственный планировщик.

  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!"));