Mono.share() allow a stream to be canceled

[rgarciapariente]

We are using Mono::share and if the mono recive a TimeOut exception the mono execution is canceled

Expected Behavior

I expect that a timeout don't cancel a Mono return by share() method.

Actual Behavior

The mono is cancelled

Steps to Reproduce

@Test
void reproCase() {
        var stream = Mono.just("Hola")
                .doOnNext(next -> System.out.println("Learning to say hi: " + next))
                .delayElement(Duration.of(5, ChronoUnit.SECONDS))
                .doOnNext(next -> System.out.println("Say hi: " + next))
                .share();

        System.out.println("Init ...");

        var result = stream
                .subscribeOn(Schedulers.boundedElastic())
                .timeout(Duration.of(1, ChronoUnit.SECONDS))
                .onErrorReturn(TimeoutException.class, "Timeout :(")
                .block();

        System.out.println("Result: " + result);

        stream.subscribe();

        Thread.sleep(10_000);
}

Result:

Init ...
Learning to say hi: Hola
Result: Timeout :(
Learning to say hi: Hola
Say hi: Hola

Note the two "Learning to say ...."

but

@Test
void workingCase() {
        var stream = Mono.just("Hola")
                .doOnNext(next -> System.out.println("Learning to say hi: " + next))
                .delayElement(Duration.of(5, ChronoUnit.SECONDS))
                .doOnNext(next -> System.out.println("Say hi: " + next))
                .share();

        System.out.println("Init ...");

        stream.subscribe();

        var result = stream
                .subscribeOn(Schedulers.boundedElastic())
                .timeout(Duration.of(1, ChronoUnit.SECONDS))
                .onErrorReturn(TimeoutException.class, "Timeout :(")
                .block();

        System.out.println("Result: " + result);

        Thread.sleep(10_000);
}

Result:

Init ...
Learning to say hi: Hola
Result: Timeout :(
Say hi: Hola

Your Environment

• Reactor version(s) used: 3.4.30
• JVM version (java -version): Java version: 17.0.4, vendor: GraalVM Community,

[chemicL]

Hello, @rgarciapariente 👋

I am not sure that I follow.

The first example looks correct

• The first subscriber cancels its processing upon timeout and doesn't consume the result.
• The second subscriber comes in next and extracts the result from the shared instance without issue.

This goes according to the javadoc:

Prepare a Mono which shares this Mono result (...)

The second example has only one subscriber, which cancels itself upon timeout. However, meanwhile, the result is delivered from the source to the subscription – which is not cancelled, and can be used again by late subscribers.

To me it looks as if the behaviour is correct. Can you please explain what seems to be the incorrect behaviour?

[chemicL]

Ah. I think I understand what you mean. You're referring to the fact that the source is being reached twice - that the "Learning to say hi: Hola" is printed out twice. Am I correct? That doesn't mean cancellation I believe, but potentially re-subscribing to the source. Is that the actual issue?

[rgarciapariente]

Many thanks for your fast answer :)

yes, that is. I am interpreting of javadoc "It's worth noting this is an un-cancellable" as meaning that "Learning to say hi: Hello" should not be executed twice in the first sample

[chemicL]

I see. Well, this is quite interesting. Indeed, the upstream subscription gets cancelled in case all downstream subscribers are gone.

A recent change: #2756.

Background: #2680 -> the same reasoning as you brought up.

It makes sense to me after looking into the background and the implementation. The current behaviour sounds like something expected. The issue I see now is the misleading documentation. As you can see in the PR, all other javadocs have been aligned, except this one. Probably an omission.

I'll mark this issue as a documentation problem. Is that ok for you @rgarciapariente ?

[rgarciapariente]

I understand and share your argument. Thank you very much for your time