Mono.share() is canceled on subscription disposal

[yamass]

Expected Behavior

When doing var sharedMono = innerMono.share(), the innerMono should not be canceled if calling sharedMono.subscribe().dispose().

At least, that's what the JavaDoc for Mono.share() suggests: It's worth noting this is an un-cancellable Subscription.

Actual Behavior

The innerMono is canceled.

Steps to Reproduce

		Mono.create(monoSink -> {
			monoSink.onCancel(() -> {
				System.out.println("CANCELED");
			});
			monoSink.success("asdf");
		})
				.subscribeOn(Schedulers.boundedElastic())
				.share()
				.subscribe()
				.dispose();

		Thread.sleep(1000);

Prints CANCELED.

By the way, innerMono.cache().subscribe().dispose does NOT cancel innerMono, although the JavaDoc makes no statement about cancellation behavior... (just replace share() with cache() in the example.

Possible Solution

This might be a bug in the implementation or documentation, or simply me not understanding the documentation right.

Please also note that the marble diagram for share() needs a fix, too. (see its cancel() arrow)

Also, please consider documenting the cancellation behavior of cache().

Meanwhile, I will use cache() as a workaround. It might actually also make sense to document the difference between these two methods.

Thanks a lot!

Your Environment

• Reactor version(s) used: 3.4.5
• JVM version (java -version): 14.0.1
• OS and version (eg uname -a): MacOS 11.2.3

[simonbasle]

we need to backtrack a bit, ensure which behavior is the "correct" one and better document the two operators behavior in face of cancellation, indeed.

[simonbasle]

if we look at Flux#share() (which is actually flux.publish().refCount()), it does cancel when going from 1 subscriber down to 0 subscriber. Flux#cache() is flux.replay().autoConnect(), which doesn't cancel on subscriber removal.

but Mono is in another place entirely. Underneath the cover, share() is implemented via NextProcessor, which usually doesn't react to cancellations of subscribers EXCEPT the subscribe() one (the dumbed down version without any handlers).

actually, there is a special case in place for mono.share().subscribe() that returns the NextProcessor as the Disposable. So calling dispose() here actually terminates the underlying NextProcessor and cancels its source subscription if it has one.

That means that calling .subscribe().dispose() on the same instance of mono.share() twice will actually ignore that second subscription. it can be better illustrated by doing:

Mono<Integer> mono = mono.just(1).share();
mono.subscribe().dispose();
mono.block();

The block() throws a java.util.concurrent.CancellationException: Disposed.

We have to re-evaluate all of these internals :(

[simonbasle]

I'm not sure what to do here in terms of timeline / milestone for changing this.
Can we consider that Flux.share().subscribe() returning a Disposable that is the same for all subsequent subscriptions() and effectively terminates the underlying processor a bug?

Opened #2756 to explore a potential fix