Spring webflux convert flux.bufferTimeout(maxSize, duration) which gives Flux<List<T>> to flux.windowTimeout(maxSize, duration).concatMap()

What I am trying to achieve:

Spring webflux and reactor offer a .bufferTimeout(maxSize, duration) operator, which gives a Flux<List<T>> on a Flux<T>

I am trying to achieve the same, not by using .bufferTimeout(maxSize, duration), but using flux.windowTimeout(maxSize, duration).concatMap()

What did I try:

I tried using concatMap as follow: flux.windowTimeout(maxSize, duration).concatMap(f -> f)

Issue:

With the above code, I am getting a T, but I would like to get the same as buffer time out.

Question:

What is the correct combination of windowTimeout + concatMap which is equivalent to buffer time out?

To achieve the same behavior as bufferTimeout(maxSize, duration) using windowTimeout(maxSize, duration) and concatMap(), you can modify your code as follows:

flux.windowTimeout(maxSize, duration)
    .concatMap(f -> f.collectList())

Explanation:

• windowTimeout(maxSize, duration) will create windows of elements based on the specified maxSize and duration.
• concatMap() will subscribe to each window as a separate Flux and emit their elements sequentially.
• By calling collectList() on each window Flux, you can transform each window into a single List emission, which will be equivalent to the behavior of bufferTimeout(maxSize, duration).
• This modification ensures that instead of emitting individual elements, you receive a Flux<List> containing the buffered elements after the specified maxSize or duration.

Please note that in this approach, each emitted List will contain the elements collected within the specified window, which may differ from the behavior of bufferTimeout(maxSize, duration) if the source Flux emits elements irregularly.

Let me know if you need further assistance!