Reactor-core: windowUntil() doesn't complete if prefetch is too small

[mikeharder]

Expected Behavior

windowUntil() should complete regardless of the prefetch value. According to the documentation:

Prefetch is a way to tune the initial request made on these inner sequences.

My interpretation is that prefetch can be used to tune performance or order of operations, but it shouldn't cause significant changes in behavior.

Actual Behavior

windowUntil() doesn't complete if prefetch is too small. Therefore, if windowUntil() needs to correctly handle arbitrary inputs, I believe prefetch should always be set to unbounded (Integer.MAX_VALUE).

Steps to Reproduce

If prefetch is large enough or unbounded (Integer.MAX_VALUE), this code generates 4 WindowFlux and then completes:

Flux.range(1,20)
    .windowUntil(i -> (i % 5 == 0), false, Integer.MAX_VALUE)
    .log()
    .subscribe();

onSubscribe([Fuseable] FluxWindowPredicate.WindowPredicateMain)
request(unbounded)
onNext(WindowFlux)
onNext(WindowFlux)
onNext(WindowFlux)
onNext(WindowFlux)
onComplete()

However, if prefetch is too small, this code will print less than 4 windows and never completes:

Flux.range(1,20)
    .windowUntil(i -> (i % 5 == 0), false, 4)
    .log()
    .subscribe();

onSubscribe([Fuseable] FluxWindowPredicate.WindowPredicateMain)
request(unbounded)
onNext(WindowFlux)

Possible Solution

Your Environment

• Reactor version(s) used: Dysprosium-SR2
• Other relevant libraries versions (eg. netty, ...):
• JVM version (javar -version): OpenJDK Runtime Environment AdoptOpenJDK (build 11.0.5+10)
• OS and version (eg uname -a): Windows 10 Version 1909

[bsideup]

(Status update)

I played a little bit with the code and noticed something:

Flux.range(1,20)
    .windowUntil(i -> (i % 5 == 0), false, 4)
    .blockLast(Duration.ofSeconds(5));

This indeed does not work (as reported).

But, if the window is subscribed, it works fine:

Flux.range(1,20)
    .windowUntil(i -> (i % 5 == 0), false, 4)
    .flatMap(it -> it)
    .blockLast(Duration.ofSeconds(5));

After some debugging I figured that this condition makes it ignore the last prefetched item and does not trigger another prefetch:
https://github.com/reactor/reactor-core/blob/f70567440236b5ca4f4ad63697a4cc9ddeac838c/reactor-core/src/main/java/reactor/core/publisher/FluxWindowPredicate.java#L717-L719

[bsideup]

Update:

As was correctly pointed out by @simonbasle, one has to subscribe on the returned windows.

However, after looking at it, we also agreed that prefetch does not make a lot of sense here and are considering to deprecate the parameter now.

[mikeharder]

Can you please clarify what is meant by "subscribe on the returned windows"? Is this code incorrect?

Flux.range(1,20)
    .windowUntil(i -> (i % 5 == 0), false, 4)
    .log()
    .subscribe();

[bsideup]

@mikeharder it is incorrect, because you never subscribe to the window. Adding .flatMap(it -> it) after windowUntil (or log) would do the job.

[mikeharder]

I don't think flatMap() is correct for my scenario, since I want to preserve the windows rather than flatten them.

I can confirm that subscribing on the returned windows causes windowUntil() to complete. Is the requirement to subscribe on the returned windows documented somewhere? I didn't see it in the documentation for windowUntil().

Also, is it only required to subscribe on the returned windows when prefetch is too small? If so, does this mean if you want to generate all the windows without subscribing on them, then prefetch should be set to Integer.MAX_VALUE? I know this works with the current version of Reactor, but I want to verify that this behavior is by design, rather than unspecified behavior that might change in the future.

Flux.range(1,20)
    .windowUntil(i -> (i % 5 == 0), false, 4)
    .log()
    .subscribe(w -> w.subscribe());

onSubscribe([Fuseable] FluxWindowPredicate.WindowPredicateMain)
request(unbounded)
onNext(WindowFlux)
onNext(WindowFlux)
onNext(WindowFlux)
onNext(WindowFlux)
onComplete()

[simonbasle]

Windows like the WindowFlux object are NOT intended to be used in isolation. They instead represent a partial view of a live sequence, and thus must be acted upon and consumed _live_. Otherwise what is the sense of windowing in the first place?

If you don't want to act upon the sub-sequence as its elements come in, you are rather in need of a buffer()ing operator.

Here the prefetch can happen to make it work because it sort of emulates requests made on each of the windows: a prefetch of 5 would request enough elements from the source to close the first window and open the next one, so you'd get 2 onNext(WindowFlux). A prefetch of Integer.MAX_VALUE requests the whole source, so the 4 windows get opened and completed.

windowUntil is a bit special because it guarantees contiguous windows, and only one active window at a time. So a WindowFlux translates individual requests to upstream replenishment on a 1-to-1 basis, so it doesn't request an excessive amount from upstream. It also tries to accommodate a case where each window is consumed in locksteps (ie. requesting one element from the currently open window then pausing shouldn't request more than one element from the source).

To answer your question, by design one should ALWAYS subscribe to a window. prefetch here is just making things confusing, so it might go away in a future release. Better rely on the common denominator of subscribing/requesting all window operators.