concatMapIterable could be enhanced to discard iterator remainder

[simonbasle]

From #1925:

I can't verify because the step before bufferUntil is concatMapIterable which produces a collection of allocated items and those don't seem to pass through doOnDiscard in case of a downstream error. Could there be an issue with concatMapIterable?

Here is a simplified test:

@Test
void concatMapIterableDoOnDiscardTest() {

	Foo foo1 = new Foo();
	Foo foo2 = new Foo();
	Foo foo3 = new Foo();

	Flux<Foo> source = Flux.just(1)
			.concatMapIterable(i -> Arrays.asList(foo1, foo2, foo3))
			.doOnDiscard(Foo.class, Foo::release);

	StepVerifier.create(source)
			.consumeNextWith(foo -> {
				foo.release();
			})
			.thenCancel()
			.verify();

	assertThat(foo1.getRefCount()).isEqualTo(0); // okay
	assertThat(foo2.getRefCount()).isEqualTo(0); // fails
	assertThat(foo3.getRefCount()).isEqualTo(0); // fails
}

static class Foo {

	int refCount = 1;

	public int getRefCount() {
		return this.refCount;
	}

	public void release() {
		this.refCount = 0;
	}
}

Originally posted by @rstoyanchev in #1925 (comment)

[simonbasle]

Currently concatMapIterable doesn't attempt to walk the remainder of the current iterator to discard further elements upon cancellation.

We could attempt to discard the iterator on top of the prefetched source elements.

This is tricky though, since an arbitrary Iterable/Iterator can be lazy and infinite...

In order to implement a best effort solution, we can retain the Class<Iterable> and inspect it, only attempting to discard the remainder of the Iterator if original iterable is either a Collection or Tuple2.

[bsideup]

Something to consider:
https://docs.oracle.com/javase/8/docs/api/java/util/Spliterator.html#trySplit--

[simonbasle]

the estimateSize() from Spliterator might be a perfect way of avoiding the infinite remainder case. We could switch the internal implementation from an Iterator-based one to a Spliterato-based one.

[rstoyanchev]

What would that mean for the API? Currently it takes a mapper with Iterable.

[bsideup]

Every Iterable can be converted into Spliterator:

https://docs.oracle.com/javase/8/docs/api/java/lang/Iterable.html#spliterator--

[simonbasle]

See more details in #2021, but going Spliterator the whole way doesn't work terribly well. For instance, the default Iterable#spliterator() creates a Spliterator over the Iterator, but its estimateSize() doesn't reflect hasNext() == false (always Long.MAX_VALUE), which is a test that we absolutely need for correct termination of the Flux.

An approach mixing the Iterator consumption of the iterable with a peek at the Spliterator#getExactSizeIfKnown() works better, at least the later won't give us false positive on which iterator we can "drain" and discard in case of cancellation.