flatMapIterable stops emmitting when published twice

[lbovet]

The following code should issue 600

    just(range(0, 300).toIterable(), range(0, 300).toIterable())
            .flatMapIterable(x->x)
            .share()
            .share()
            .count()
            .log()
            .subscribe();

But it issues 256

Whereas

    just(range(0, 300).toIterable(), range(0, 300).toIterable())
            .flatMap(Flux::fromIterable)
            .share()
            .share()
            .count()
            .log()
            .subscribe();

works as expected

Note: I could produce the wrong behaviour only with multiple iterables in the "just" and publishing twice.

[lhotari]

Failing test:

	/**
	 * See https://github.com/reactor/reactor-core/issues/508
	 */
	@Test
	public void testPublishingTwice() {
		StepVerifier.create(Flux.just(range(0, 300).toIterable(), range(0, 300).toIterable())
				.flatMapIterable(x->x)
				.share()
				.share()
				.count())
				.expectNext(600L)
				.verifyComplete();
	}

Passes with this modification:

	/**
	 * See https://github.com/reactor/reactor-core/issues/508
	 */
	@Test
	public void testPublishingTwice() {
		StepVerifier.create(Flux.just(range(0, 300).toIterable(), range(0, 300).toIterable())
				.flatMapIterable(x->x)
				.share()
				.publish(600).refCount() // this makes it pass
				.count())
				.expectNext(600L)
				.verifyComplete();
	}

[lhotari]

I guess it's related to the backpressure behaviour of publish.

Backpressure will be coordinated on Subscription#request and if any Subscriber is missing demand (requested = 0), multicast will pause pushing/pulling.

[lhotari]

I made PR #509 with a fix for this issue.