Spring WebFlux-将磁通转换为List <Object>

如何解决Spring WebFlux-将磁通转换为List <Object>

我正在学习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);
        User stock3 = new User(13,"ccc",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的评论，我尝试了此操作：

@RequestMapping(value = "/livelistofusersall",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秒后加载，但“仅一次”。我在这里想念什么？

解决方法

您可以在Flux中使用collectList()运算符，以给出列表的单声道。

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