I have a case where I'm attempting to dispose a connection via an RSocketRequester using the .dispose() method given a particular condition. In version 1.1.3 this worked no problem. However, in version 1.1.4 these changes were made which are now keeping the connection open after attempting to dispose. I can see that a ConnectionErrorException is sent by the requester, however, the sender continues to send frames after this. I believe onClose is never triggered as the tryShutdown is never attempted inside the RSocketRequester.

Expected Behavior

I would expect that when calling .dispose via the RSocketRequester, the connection would terminate.

Actual Behavior

Connection remains open.

Steps to Reproduce

rSocketRequester
            .route(<routeToStream)
            .data(<data>)
            .retrieveFlux(<datatype>)
            .filter(<conditional>)
            .doOnNext(value -> {
                policyBrokerConnector.dispose();
            })

Possible Solution

Your Environment

• RSocket version(s) used: 1.1.4
• Other relevant libraries versions (eg. netty, ...):
• Platform (eg. JVM version (javar -version) or Node version (node --version)):
• OS and version (eg uname -a):