fix #2014 Discard concatMapIterable/fromIterable's remainder on Cancel

In this change, the goal is to discard elements of the Iterable that
haven't been processed yet. The challenge is to avoid attempting doing
so for _infinite_ Iterables (which would lead to infinite discarding
loops).

If the Iterable is a Collection, it should be finite.

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.

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).

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

Author: simonbasle

reactor-core/src/main/java/reactor/core/publisher/Flux.java

@@ -30,6 +30,7 @@
 import java.util.Objects;
 import java.util.Queue;
 import java.util.Set;
+import java.util.Spliterator;
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -3654,7 +3655,9 @@ public final <V> Flux<V> concatMapDelayError(Function<? super T, ? extends Publi
 	 * Thus {@code flatMapIterable} and {@code concatMapIterable} are equivalent offered as a discoverability
 	 * improvement for users that explore the API with the concat vs flatMap expectation.
 	 *
-	 * @reactor.discard This operator discards elements it internally queued for backpressure upon cancellation.
+	 * @reactor.discard Upon cancellation, this operator discards {@code T} elements it prefetched and, in
+	 * some cases, attempts to discard remainder of the currently processed {@link Iterable} (if it can
+	 * safely assume the iterator is not infinite, see {@link Spliterator#getExactSizeIfKnown()}).
 	 *
 	 * @param mapper the {@link Function} to transform input sequence into N {@link Iterable}
 	 * @param <R> the merged output sequence type
@@ -3677,7 +3680,9 @@ public final <R> Flux<R> concatMapIterable(Function<? super T, ? extends Iterabl
 	 * Thus {@code flatMapIterable} and {@code concatMapIterable} are equivalent offered as a discoverability
 	 * improvement for users that explore the API with the concat vs flatMap expectation.
 	 *
-	 * @reactor.discard This operator discards elements it internally queued for backpressure upon cancellation.
+	 * @reactor.discard Upon cancellation, this operator discards {@code T} elements it prefetched and, in
+	 * some cases, attempts to discard remainder of the currently processed {@link Iterable} (if it can
+	 * safely assume the iterator is not infinite, see {@link Spliterator#getExactSizeIfKnown()}).
 	 *
 	 * @param mapper the {@link Function} to transform input sequence into N {@link Iterable}
 	 * @param prefetch the maximum in-flight elements from each inner {@link Iterable} sequence
@@ -4925,7 +4930,9 @@ public final <R> Flux<R> flatMap(
 	 * Thus {@code flatMapIterable} and {@code concatMapIterable} are equivalent offered as a discoverability
 	 * improvement for users that explore the API with the concat vs flatMap expectation.
 	 *
-	 * @reactor.discard This operator discards elements internally queued for backpressure upon cancellation or error triggered by a data signal.
+	 * @reactor.discard Upon cancellation, this operator discards {@code T} elements it prefetched and, in
+	 * some cases, attempts to discard remainder of the currently processed {@link Iterable} (if it can
+	 * safely assume iterator is not infinite, see {@link Spliterator#getExactSizeIfKnown()}).
 	 *
 	 * @param mapper the {@link Function} to transform input sequence into N {@link Iterable}
 	 * @param <R> the merged output sequence type
@@ -4949,7 +4956,9 @@ public final <R> Flux<R> flatMapIterable(Function<? super T, ? extends Iterable<
 	 * Thus {@code flatMapIterable} and {@code concatMapIterable} are equivalent offered as a discoverability
 	 * improvement for users that explore the API with the concat vs flatMap expectation.
 	 *
-	 * @reactor.discard This operator discards elements internally queued for backpressure upon cancellation or error triggered by a data signal.
+	 * @reactor.discard Upon cancellation, this operator discards {@code T} elements it prefetched and, in
+	 * some cases, attempts to discard remainder of the currently processed {@link Iterable} (if it can
+	 * safely assume the iterator is not infinite, see {@link Spliterator#getExactSizeIfKnown()}).
 	 *
 	 * @param mapper the {@link Function} to transform input sequence into N {@link Iterable}
 	 * @param prefetch the maximum in-flight elements from each inner {@link Iterable} sequence