Optimize for Flux to/from Mono conversions by Reactor [SPR-17203]

[spring-projects-issues]

Kamil opened SPR-17203 and commented

I'm providing test case

If you run $ ./gradlew test then test will fail with:

java.lang.AssertionError: PublisherProbe should not have been cancelled but it was
	at reactor.test.publisher.PublisherProbe.assertWasNotCancelled(PublisherProbe.java:73)
	at com.example.demo.MyControllerTests.shouldAllowRetrieval(MyControllerTests.java:47)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
	at java.lang.reflect.Method.invoke(Unknown Source)
	at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:436)
	at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:115) 

but if you edit build.gradle file and change Spring version back to 5.0.7.RELEASE (comment line 18) then test will pass.

 

It is believed that this commit broke it: d1c9401#diff-5d32a8109e4c3e46a7a012dfca5c0384R109

---

Affects: 5.0.8

Attachments:

• spring_reactor_bug.zip (58.10 kB)

Referenced from: commits 1bdbc7b

1 votes, 5 watchers

[spring-projects-issues]

Brian Clozel commented

According to Simon's comment on the Spring Boot issue, this behavior is perfectly in line with Reactor internals. So your test case is wrong in the first place; this is why I've change the title of this issue and made it an improvement.

The reactor team might work on an optimization feature that will detect when Flux and Mono instances are converted to and from repeatedly and make those calls no-ops when the reactive pipeline is assembled.

This improvement is about making those cycles more obvious and avoid chaining another operator between the two operations, when possible. This would mean, for example, converting the following code in AbstractJackson2Encoder:

public Flux<DataBuffer> encode(Publisher inputstream) {

    if (inputStream instanceof Mono) {

        Flux<DataBuffer> result = Flux.from(inputStream).map(value -> encodeValue(...));
        // should become
        Flux<DataBuffer> result = Mono.from(inputStream).map(value -> encodeValue(...)).flux();
        return result;
    }

}

Because later, we have:

Flux<DataBuffer> body = encode(inputstream);
if (inputStream instanceof Mono) {
    Mono.from(body)...
}

This effectively makes a Mono.from(mono.flux()), which can be detected.
So as a general rule, we should make the Mono / Flux conversions at the limits of a method: right when they're introduced as parameters, or right before they're about to be returned.

This issue is about finding and fixing those as much as possible in our codebase.

 

[spring-projects-issues]

Kamil commented

But I don't have any Mono->Flux conversion in my code.

And the question is: "why it worked in Spring 5.0.7" ?

[spring-projects-issues]

Simon Baslé commented

Kamil the Mono to Flux and back to Mono conversion is in Spring's code, part of an improvement for computing Content-Length header. But in your test the cancel assertion is not really relevant.

The only caveat with this issue is when the original Mono has a .doOnTerminate() callback, as this one gets short-circuited by the cancel. Note that other callbacks like doOnSuccess  or doOnError (for a failing mono) should work fine.

[spring-projects-issues]

Simon Baslé commented

Brian Clozel do you see a way to avoid that Mono.from on the Flux<DataBuffer> in the code? I guess other encoders could also convert a Flux to Mono, right?