Spring MessagingRSocket incorrectly tries to release entire input stream on channel interaction

The issue was discovered in Spring Security tests after a recent fix in Reactor Core to the switchOnFirst operator which now properly detects and flags the scenario.

In an RSocket channel interaction, a Spring server responder fails to find a matching route and in that scenarios the exception occurs. The server handles this with switchOnFirst and in the end, if input stream wasn't subscribed to (e.g. because there was no matching handler), it subscribes in order to release the payloads resulting in:

reactor.core.Exceptions$ErrorCallbackNotImplemented: java.util.concurrent.CancellationException: FluxSwitchOnFirst has already been cancelled
Caused by: java.util.concurrent.CancellationException: FluxSwitchOnFirst has already been cancelled
    at reactor.core.publisher.FluxSwitchOnFirst$SwitchOnFirstMain.subscribe(FluxSwitchOnFirst.java:348) ~[reactor-core-3.3.4.BUILD-SNAPSHOT.jar:3.3.4.BUILD-SNAPSHOT]
    at reactor.core.publisher.Flux.subscribe(Flux.java:8260) ~[reactor-core-3.3.4.BUILD-SNAPSHOT.jar:3.3.4.BUILD-SNAPSHOT]
    at reactor.core.publisher.Flux.subscribeWith(Flux.java:8424) ~[reactor-core-3.3.4.BUILD-SNAPSHOT.jar:3.3.4.BUILD-SNAPSHOT]
    at reactor.core.publisher.Flux.subscribe(Flux.java:8231) ~[reactor-core-3.3.4.BUILD-SNAPSHOT.jar:3.3.4.BUILD-SNAPSHOT]
    at reactor.core.publisher.Flux.subscribe(Flux.java:8158) ~[reactor-core-3.3.4.BUILD-SNAPSHOT.jar:3.3.4.BUILD-SNAPSHOT]
    at reactor.core.publisher.Flux.subscribe(Flux.java:8101) ~[reactor-core-3.3.4.BUILD-SNAPSHOT.jar:3.3.4.BUILD-SNAPSHOT]
    at org.springframework.messaging.rsocket.annotation.support.MessagingRSocket.lambda$handleAndReply$5(MessagingRSocket.java:177) ~[spring-messaging-5.2.4.RELEASE.jar:5.2.4.RELEASE]

We shouldn't subscribe to the entire input which in a channel interaction can be quite long. In any case only the first payload is received at that point, and it's the only one that needs releasing.