在while循环中构建通量对象

Question

我有一个有趣的问题，我不知道如何在不调用block（）方法的情况下解决。

>

我的方法接收用户作为参数。

它调用外部服务并接收Mono

如果Mono不包含错误和user.getDepartment（）. startsTo（"Development"）我想将其添加到Flux

通量应该是我的方法的结果

现在，不幸的是，在调用我收到的Mono的＜code＞block（）-方法之前，我应该从列表中初始化Flux：

  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

通过使用

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

//or

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

Answer 2

您可以使用递归应用getManager函数并将结果组合到Flux中的Mono#扩展

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