在 while 循环中构建一个 Flux 对象

Question

我有一个有趣的问题，如果不调用方法我不知道如何解决block()。

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

2. 它调用外部服务并接收 Mono

3. 如果 Mono 不包含错误并且user.getDepartment().startsWith("Development")我想将其添加到 Flux

4. Flux 应该是我方法的结果

block()不幸的是，现在我应该在调用我收到的 Mono的 -method 之前从列表中初始化 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

您可以使用Mono#expand递归应用getManager函数并将结果合并到Flux<User>：Flux<User> getUsers(User user) {&nbsp; return getManager(user).expand(manager -> getManager(manager));}Mono<User> getManager(User user) {&nbsp; if (user.containsManager()) {&nbsp; &nbsp; return externalUserServiceClient.getManager(user)&nbsp; &nbsp; &nbsp; .filter(manager -> manager.getDepartment().startsWith("Development"));&nbsp; } else {&nbsp; &nbsp; return Mono.empty();&nbsp; }}

Answer 2

通过使用externalUserServiceClient.getManager(user).map(manager -> ...);//orexternalUserServiceClient.getManager(user).flatMap(manager -> ...);