Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use RSocketRequester.rsocketClient() to connect

Browse Source 
Related to https://github.com/spring-projects/spring-framework/issues/25332

* Delegate subscription (and therefore auto-connection) into underlying `RSocketClient`
in the `ClientRSocketConnector`
This commit is contained in:
Artem Bilan
2020-07-22 08:58:11 -04:00
parent 40b0031e0f
commit ff44382d94
2 changed files with 24 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java
View File

@@ -26,12 +26,11 @@ import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
import io.rsocket.core.RSocketConnector;
import io.rsocket.transport.ClientTransport;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.transport.netty.client.WebsocketClientTransport;
import reactor.core.Disposable;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
/**
* A client {@link AbstractRSocketConnector} extension to the RSocket connection.
@@ -40,7 +39,7 @@ import reactor.core.publisher.Sinks;
*
* @since 5.2
*
* @see io.rsocket.RSocketFactory.ClientRSocketFactory
* @see io.rsocket.core.RSocketConnector
* @see RSocketRequester
*/
public class ClientRSocketConnector extends AbstractRSocketConnector {
@@ -176,27 +175,21 @@ public class ClientRSocketConnector extends AbstractRSocketConnector {
public void afterPropertiesSet() {
super.afterPropertiesSet();
Sinks.StandaloneMonoSink<RSocketConnector> rsocketConnector = Sinks.promise();
RSocketRequester rsocketRequester =
RSocketRequester.builder()
.dataMimeType(getDataMimeType())
.metadataMimeType(getMetadataMimeType())
.rsocketStrategies(getRSocketStrategies())
.setupData(this.setupData)
.setupRoute(this.setupRoute, this.setupRouteVars)
.apply((builder) -> this.setupMetadata.forEach(builder::setupMetadata))
.rsocketConnector(this.connectorConfigurer)
.rsocketConnector((connector) -> {
connector.acceptor(this.rSocketMessageHandler.responder());
rsocketConnector.success(connector);
})
.transport(this.clientTransport);
RSocketRequester rsocketRequester = RSocketRequester.builder()
.dataMimeType(getDataMimeType())
.metadataMimeType(getMetadataMimeType())
.rsocketStrategies(getRSocketStrategies())
.setupData(this.setupData)
.setupRoute(this.setupRoute, this.setupRouteVars)
.rsocketConnector(this.connectorConfigurer)
.rsocketConnector((connector) ->
connector.acceptor(this.rSocketMessageHandler.responder()))
.apply((builder) -> this.setupMetadata.forEach(builder::setupMetadata))
.transport(this.clientTransport);
this.rsocketRequesterMono =
rsocketConnector.asMono()
.flatMap(rSocketConnector -> rSocketConnector.connect(this.clientTransport))
.thenReturn(rsocketRequester)
Mono.just(rsocketRequester)
.doOnSubscribe((sub) -> rsocketRequester.rsocketClient().source().subscribe())
.cache();
}
@@ -215,7 +208,8 @@ public class ClientRSocketConnector extends AbstractRSocketConnector {
@Override
public void destroy() {
this.rsocketRequesterMono
.doOnNext(RSocketRequester::dispose)
.flatMap((requester) -> requester.rsocketClient().source())
.doOnNext(Disposable::dispose)
.subscribe();
}

22
spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java
View File

@@ -59,10 +59,8 @@ import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import io.rsocket.RSocket;
import io.rsocket.core.RSocketConnector;
import io.rsocket.core.RSocketServer;
import io.rsocket.frame.decoder.PayloadDecoder;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.transport.netty.server.CloseableChannel;
import io.rsocket.transport.netty.server.TcpServerTransport;
import reactor.core.Disposable;
@@ -499,21 +497,15 @@ public class RSocketOutboundGatewayIntegrationTests {
@Bean(destroyMethod = "dispose")
@Nullable
public RSocket rsocketForServerRequests() {
Sinks.StandaloneMonoSink<RSocketConnector> rsocketConnector = Sinks.promise();
RSocketRequester.builder()
return RSocketRequester.builder()
.setupRoute("clientConnect")
.rsocketConnector(connector -> {
connector.acceptor(
RSocketMessageHandler.responder(RSocketStrategies.create(), controller()));
rsocketConnector.success(connector);
})
.tcp("localhost", server.address().getPort());
return rsocketConnector.asMono()
.flatMap(rSocketConnector ->
rSocketConnector.connect(
TcpClientTransport.create("localhost", server.address().getPort())))
.rsocketConnector(connector ->
connector.acceptor(
RSocketMessageHandler.responder(RSocketStrategies.create(), controller())))
.tcp("localhost", server.address().getPort())
.rsocketClient()
.source()
.block();
}