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

Question

我有一个有趣的问题，如果不调用方法我不知道如何解决block()。我的方法接收用户作为参数。它调用外部服务并接收 Mono如果 Mono 不包含错误并且user.getDepartment().startsWith("Development")我想将其添加到 FluxFlux 应该是我方法的结果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) {

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