fix #2014 Cancel discards concatMapIterable/fromIterable's remainder …

[simonbasle]

…from iterator

If both an Iterator and a Spliterator can be generated for each of the
processed Iterables, then the Spliterator is used to ensure the Iterable
is SIZED. This allows us to safely assume we can iterate over the
remainder of the iterator when cancelling, in order to discard its
elements that weren't emitted.

Not doing this check would likely cause trouble with infinite discarding
loops in the case of infinite Iterables (which is technically possible).

For Streams, since both the iterator() and spliterator() methods are
terminating the Stream we only generate the Spliterator. We use it to
check SIZED and then wrap it in an Iterator adapter for iteration (which
is what BaseStream does by default).

Note that using a Spliterator to drive the internal iteration doesn't
work that well, since the default Iterable#spliterator isn't SIZED and
its estimatedSize() method doesn't behave like hasNext().
Iterator#hasNext is far better suited for looking ahead of the emitted
element to trigger onComplete immediately after the last onNext.

[codecov-io]

Codecov Report

Merging #2021 into 3.2.x will increase coverage by 0.12%.
The diff coverage is 85.92%.

@@             Coverage Diff              @@
##              3.2.x    #2021      +/-   ##
============================================
+ Coverage     84.61%   84.73%   +0.12%     
- Complexity     3967     3981      +14     
============================================
  Files           364      364              
  Lines         30178    30260      +82     
  Branches       5612     5614       +2     
============================================
+ Hits          25536    25642     +106     
+ Misses         3032     3010      -22     
+ Partials       1610     1608       -2

Impacted Files	Coverage Δ	Complexity Δ
...ore/src/main/java/reactor/core/publisher/Mono.java	94.49% <ø> (ø)	268 <0> (ø)	⬇️
...ore/src/main/java/reactor/core/publisher/Flux.java	98.63% <ø> (ø)	517 <0> (ø)	⬇️
...va/reactor/core/publisher/MonoFlattenIterable.java	53.33% <100%> (+29.19%)	3 <0> (+1)	⬆️
...c/main/java/reactor/core/publisher/FluxStream.java	94.44% <100%> (+0.69%)	2 <0> (ø)	⬇️
...va/reactor/core/publisher/FluxFlattenIterable.java	84.69% <76.05%> (+1.65%)	6 <0> (ø)	⬇️
...main/java/reactor/core/publisher/FluxIterable.java	77.43% <93.75%> (+0.28%)	13 <5> (+1)	⬆️
...rc/main/java/reactor/core/publisher/Operators.java	83.08% <97.61%> (+1.68%)	133 <7> (+8)	⬆️
...e/src/main/java/reactor/core/publisher/Traces.java	69.65% <0%> (ø)	25% <0%> (ø)	⬇️
...ain/java/reactor/util/concurrent/WaitStrategy.java	50.42% <0%> (-1.71%)	11% <0%> (ø)
...in/java/reactor/core/publisher/FluxWindowWhen.java	86.05% <0%> (+4.32%)	2% <0%> (ø)	⬇️
... and 22 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update c9c0ae0...f0dfc64. Read the comment docs.

[bsideup]

The following block seems to be duplicated many times:

Context context = actual.currentContext();
Operators.onDiscardQueueWithClear(queue, context, null);
Operators.onDiscardMultiple(it, itFinite, context);

WDYT about extracting it as well?

[simonbasle]

The following block seems to be duplicated many times:

Context context = actual.currentContext();
Operators.onDiscardQueueWithClear(queue, context, null);
Operators.onDiscardMultiple(it, itFinite, context);

WDYT about extracting it as well?

Not as obvious an improvement as the methods can make sense separately.

[bsideup]

@simonbasle we don't need to remove them, but "group". It feels weird to always call two "onDiscard" methods, feels like it should be a single method

[rstoyanchev]

One thing I noticed while reviewing is that Operators#onDiscard* methods put try-catch around the entire loop. So if the discard action throws an error for one item, the rest will not be discarded. Should those keep going to try to discard as many as possible?

[simonbasle]

One thing I noticed while reviewing is that Operators#onDiscard* methods put try-catch around the entire loop. So if the discard action throws an error for one item, the rest will not be discarded. Should those keep going to try to discard as many as possible?

Indeed, that is another avenue of improvement. I'll introduce a separate commit, so I'll clean up the history of this PR to allow for a rebase and merge.

[simonbasle]

@bsideup @rstoyanchev can you review this? should be the final state in terms of code, and once approved I'll locally squash the commits to get 2 separates commits (one that addresses the cancel, the other that makes doOnDiscard more resilient).

[simonbasle]

Rebased on top of 3.2.x and cleaned up the commits, will merge as soon as tests pass.

[reactorbot]

@simonbasle this PR seems to have been merged on a maintenance branch, please ensure the change is merge-forwarded to intermediate maintenance branches and up to master 🙇