Построить объект Flux в то время как цикл

Question

У меня есть интересная проблема, которую я не знаю, как решить без вызова block() метода.

1. мой метод получает пользователя в качестве аргумента.

2. вызывает внешнюю службу и получает Mono

3. , если в Mono нет ошибки и user.getDepartment().startsWith("Development") Я хочу добавить это в поток

4. Флюс должен быть результатом моего метода

На данный момент, к сожалению, я должен инициализировать Flux из списка перед вызовом block() -метода для Mono, который я получаю:

  Flux<User> getUsers(User user) {

     List<Users> developmentUsers = new ArrayList<User>();

     while (user.containsManager()) {

         val resultUser = externalUserServiceClient.getManager(user).block(); //externalUserServiceClient.getManager(user) should return a Mono<User>
         if (resultUser.getDepartment().startsWith("Development"))
             developemtnUsers.add(resultUser);
         user = resultUser;
     }

     return Flux.fromIterable(developmentUsers);
  }

Я уверен, что должен быть способ не прерывать цепочку асинхронных процессов. Ты знаешь как?

Answer 1

Вы можете использовать Mono#expand, который рекурсивно применяет функцию getManager и объединяет результаты в Flux<User>:

Flux<User> getUsers(User user) {
  return getManager(user).expand(manager -> getManager(manager));
}

Mono<User> getManager(User user) {
  if (user.containsManager()) {
    return externalUserServiceClient.getManager(user)
      .filter(manager -> manager.getDepartment().startsWith("Development"));
  } else {
    return Mono.empty();
  }
}

Answer 2

Используя

externalUserServiceClient.getManager(user).map(manager -> ...);

//or

externalUserServiceClient.getManager(user).flatMap(manager -> ...);