Rework FluxReplay to avoid hanging, but reject 0 size

[OlegDokuka]

closes #1921

This PR adds a prefetch strategy driven by active subscribers so the next prefetch happens only when all the subscribers have consumed a particular element. To enable that, now every impl has an index actively utilized to identify if specific element is a trigger for the next prefetch.

In case there are no subscribers, the prefetch strategy is self-driven and requests more when there are produced N elements into the cache

Signed-off-by: Oleh Dokuka odokuka@vmware.com

[simonbasle]

as demonstrated by (currently failing) reactor.core.publisher.SinksTest.MulticastReplayN#checkSemanticsSize0, limit can be zero here which would lead to ArithmeticException.

[OlegDokuka]

Remove tests that check zero history size and added protection against passing zero as a history size since it is meaning-less for replay operator (in such scenario we can suggest alternative represented in the .publish operator)

[OlegDokuka]

@simonbasle most of the things were fixed / improved. If all good, will be adding stresstests

[simonbasle]

a small clarification on javadoc wording still, but otherwise looks good.
the commit message will need to be explicit that this commit removes support for historySize(0), just in case some users weirdly depend on it.

go ahead with stress tests

[simonbasle]

which of SUBSCRIBED_FLAG and CONNECTED_FLAG indicates subscription reception? both this javadoc and the one below now correctly link to the right constant, but they otherwise use the exact same phrasing.

[OlegDokuka]

fixed

[simonbasle]

For the pragmatic pov of not breaking anybody that would depend on history size 0 in a patch release (probably by mistake, but still...) what do you think about the operator method delegating to Flux.cache when size is zero @OlegDokuka?

[OlegDokuka]

@sbasleTo be BC I would replace that automatically with with Flux Publish

[simonbasle]

I would replace that automatically with with Flux Publish

Yeah I think that's what I meant: Flux#replay(int size) with size == 0 would delegate to Flux#publish() (indeed, since it returns a ConnectableFlux while cache returns a Flux)

There could also be a warning logged, or a mention in the release notes that next major version will reject 0 parameter.

[simonbasle]

This is not entirely ideal, but we're making the bet here that nobody actually uses historySize == 0 in the Sinks side of things, since it is a younger API.

Still marking as warn, because a 0 value is now actively rejected by an IllegalArgumentException.

For Flux.replay(historySize), we're a bit more conservative and continue allowing that usecase by transparently switching to a FluxPublish implementation rather than FluxReplay. A future version should probably also explicitly reject such calls, however.

[osi]

This is not entirely ideal, but we're making the bet here that nobody actually uses historySize == 0 in the Sinks side of things, since it is a younger API.

While I don't remember why I had this construct, I do have a usage of

Sinks.many().replay().limit(0)

within my tests which failed when ingesting the dependency update with this change.

(I suspect I wanted a sink that would allow multiple subscriptions and replay terminal signals)

[osi]

How can I create a sink that:

• Does not fail for emissions with zero subscribers
• Allows resubscription

Using Sinks.many().replay().limit(0) did this.

None of the multicast sinks work, as one can't set the backpressure buffer size to 0. The direct* versions fail when there are no subscribers.