java - 在 Webclient Post 之前检测到空的 Flux 窗口

Question

引用我之前的问题Splitting a WebClient Post of a Streaming Flux into JSON Arrays ，我正在使用；

myFlux
 .window(5)
 .flatMap(window -> client
  .post()
  .body(window, myClass.class)
  .exchange()
  .flatMap(response -> response.bodyToMono)
 )
 .subscribe();

这很好用。但是，在速度慢的一天，5 条消息需要一段时间才能到达，并且在 window 已满之前，window 不会发送任何内容。所以 我切换到 windowTimeout(5, Duration.ofSeconds(5))。

现在，如果没有数据并且超过了 Duration，则代码将传播一个空的 window，这会导致发布一个空数组。

如何检测空窗口并且不运行post？

Answer 1

不幸的是，如果不完整读取整个 Flux，就无法知道 Flux 将发出多少个项目。

由于您的窗口尺寸相对较小，您可以使用 .collectList() 将 Flux 发出的所有项目收集到 List 中，然后检查是否有在发送请求之前列表为空。

myFlux
    .windowTimeout(5, Duration.ofSeconds(5))
    .flatMap(window ->
        // collect everything in the window into a list
        window.collectList()
             // ignore empty windows
            .filter(list -> !list.isEmpty())
             // send the request
            .flatMap(list -> client
                .post()
                .body(Flux.fromIterable(list), MyClass.class)
                .exchange()
                .flatMap(response -> response.bodyToMono(MyResponse.class))))