Spring WebFlux - 将 Flux 转换为列表<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);
     }

到目前为止一切都很好。我能够 return 每 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", 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() 运算符，它给出一个单声道列表。

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

Spring WebFlux - 将 Flux 转换为列表<Object>

Spring WebFlux - Convert Flux to List<Object>

spring

h2

spring-webflux

r2dbc