Spring WebFlux - 将 Flux 转换为 List<Object>

Question

我正在学习 Spring WebFlux。

我的实体是这样的：

@Table("users")
public class User {
    @Id
    private Integer id;
    private String name;
    private int age;
    private double salary;
}

我有一个存储库（使用 H2 数据库的 R2），如下所示：

public interface UserRepository extends ReactiveCrudRepository<User,Integer> {
   
}

我的控制器是：

    @Autowired
    private UserRepository userRepository;

    private static List<User> userList = new ArrayList<>();

    @PostConstruct
    public void initializeStockObjects() {
        User stock1 = new User(11, "aaaa", 123, 123);
        User stock2 = new User(12, "bbb", 123, 123);
        User stock3 = new User(13, "ccc", 123, 123);
        userList.add(stock1);
        userList.add(stock2);
        userList.add(stock3);
    }

    @RequestMapping(value = "/livelistofusers", method = RequestMethod.GET, produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<List<User>> getUsers() {
        return getUserData(userList);
    }

    public Flux<List<User>> getUserData(List<User> userList) {
       Flux<Long> interval = Flux.interval(Duration.ofSeconds(3));
       interval.subscribe((i) -> userList.forEach(user -> addNewUser(user)));
       Flux<List<User>> transactionFlux = Flux.fromStream(Stream.generate(() -> userList));
       return Flux.zip(interval, transactionFlux).map(Tuple2::getT2);
     }

到目前为止一切顺利。我能够每 3 秒将整个用户列表返回到视图中。这里完全没有问题。

现在，我想将 Flue 即 Flux Flux2 = userRepository.findAll() 发送到视图。这意味着，除了return getUserData(userList);，我该怎么做return getUserData(flux2(...what should I do here ???... I tried couple of things but I end up making the Blocking list instead of Non-Blocking ...));？

问题：我怎样才能做到这一点？即如何每 3 秒将整个 Flux 发送到我的视图。我在这里感到迷失，毫无头绪。任何相关的帮助链接或解决方案将不胜感激。

编辑：

根据 Nipuna 的 cmets 我试过这个：

@RequestMapping(value = "/livelistofusersall", method = RequestMethod.GET, produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<List<User>>  getUsersall() {
        Flux<Long> interval = Flux.interval(Duration.ofSeconds(3));
        interval.subscribe((i) -> userRepository.findAll());
        Flux<List<User>> transactionFlux = userRepository.findAll().collectList().flatMapMany(Flux::just);
        return Flux.zip(interval, transactionFlux).map(Tuple2::getT2);
    }

但现在在我的上下文路径中，列表加载，但等待 3 秒后“仅加载一次”。我在这里缺少什么？

Answer 1

您可以在 Flux 中使用 collectList() 运算符，这会给出 List 的 Mono。

userRepository.findAll().collectList().flatMapMany(Flux::just);