java在while循环中构建Flux对象

Question

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

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

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

3. 如果Mono不包含错误和user.getDepartment().startsWith("Development")，我想将其添加到通量中

4. 流量应该是我的方法的结果

现在，在调用我收到的Mono的block()-方法之前，我应该从列表中初始化通量：

  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

可以使用^{}递归地应用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 -> ...);