From ff15d5265d0f179e30a5136512a113e044236118 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 27 Aug 2019 16:46:57 -0400 Subject: [PATCH] Align RSocket module with the latest SF * Upgrade to RSocket API `1.0.0-RC3-SNAPSHOT` * Use the same `rSocketMessageHandler.responder()` callback for client and server rsocket factories configuration * Clean up `ClientRSocketConnector` and `ServerRSocketConnector` JavaDocs not mentioning overriding any more * Use `WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA` constant instead of built-in literal * Fix `RSocketOutboundGatewayIntegrationTests` according the SF changes --- build.gradle | 2 +- .../integration/rsocket/AbstractRSocketConnector.java | 5 ++++- .../integration/rsocket/ClientRSocketConnector.java | 9 ++------- .../integration/rsocket/ServerRSocketConnector.java | 7 +------ .../outbound/RSocketOutboundGatewayIntegrationTests.java | 4 ++-- 5 files changed, 10 insertions(+), 17 deletions(-) diff --git a/build.gradle b/build.gradle index f7f987685d..36674bcac0 100644 --- a/build.gradle +++ b/build.gradle @@ -88,7 +88,7 @@ ext { reactorVersion = '3.3.0.M3' resilience4jVersion = '0.16.0' romeToolsVersion = '1.12.1' - rsocketVersion = '0.12.2-RC4' + rsocketVersion = '1.0.0-RC3-SNAPSHOT' servletApiVersion = '4.0.1' smackVersion = '4.3.4' springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.2.0.BUILD-SNAPSHOT' diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java index 9a3d99cf0e..4cd297fe3f 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java @@ -32,6 +32,8 @@ import org.springframework.util.Assert; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; +import io.rsocket.metadata.WellKnownMimeType; + /** * A base connector container for common RSocket client and server functionality. *

@@ -53,7 +55,8 @@ public abstract class AbstractRSocketConnector private MimeType dataMimeType = MimeTypeUtils.TEXT_PLAIN; - private MimeType metadataMimeType = new MimeType("message", "x.rsocket.composite-metadata.v0"); + private MimeType metadataMimeType = + MimeTypeUtils.parseMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.toString()); private RSocketStrategies rsocketStrategies = RSocketStrategies.builder() diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java index 920360e94b..d7104bcf73 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java @@ -34,12 +34,7 @@ import reactor.core.Disposable; import reactor.core.publisher.Mono; /** - * A client {@link AbstractRSocketConnector} extension to the RSocket server. - *

- * Note: the {@link RSocketFactory.ClientRSocketFactory#acceptor(java.util.function.Function)} - * in the provided {@link #factoryConfigurer} is overridden with an internal - * {@link IntegrationRSocketMessageHandler#clientResponder()} - * for the proper Spring Integration channel adapter mappings. + * A client {@link AbstractRSocketConnector} extension to the RSocket connection. * * @author Artem Bilan * @@ -126,7 +121,7 @@ public class ClientRSocketConnector extends AbstractRSocketConnector { .dataMimeType(getDataMimeType().toString()) .metadataMimeType(getMetadataMimeType().toString()); this.factoryConfigurer.accept(clientFactory); - clientFactory.acceptor(this.rSocketMessageHandler.clientResponder()); + clientFactory.acceptor(this.rSocketMessageHandler.responder()); Payload connectPayload = EmptyPayload.INSTANCE; if (this.connectRoute != null) { connectPayload = DefaultPayload.create(this.connectData, this.connectRoute); diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java index ca049cad84..4fab5a08c1 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java @@ -49,11 +49,6 @@ import reactor.netty.http.server.HttpServer; /** * A server {@link AbstractRSocketConnector} extension to accept and manage client RSocket connections. - *

- * Note: the {@link RSocketFactory.ServerRSocketFactory#acceptor(io.rsocket.SocketAcceptor)} - * in the provided {@link #factoryConfigurer} is overridden with an internal - * {@link ServerRSocketMessageHandler#serverResponder()} - * for the proper Spring Integration channel adapter mappings. * * @author Artem Bilan * @@ -131,7 +126,7 @@ public class ServerRSocketConnector extends AbstractRSocketConnector this.serverMono = serverFactory - .acceptor(serverRSocketMessageHandler().serverResponder()) + .acceptor(serverRSocketMessageHandler().responder()) .transport(this.serverTransport) .start() .cache(); diff --git a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java index 3d52642c1a..f39abf6a1a 100644 --- a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java +++ b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java @@ -119,7 +119,7 @@ public class RSocketOutboundGatewayIntegrationTests { serverContext = new AnnotationConfigApplicationContext(ServerConfig.class); server = RSocketFactory.receive() .frameDecoder(PayloadDecoder.ZERO_COPY) - .acceptor(serverContext.getBean(RSocketMessageHandler.class).serverResponder()) + .acceptor(serverContext.getBean(RSocketMessageHandler.class).responder()) .transport(TcpServerTransport.create("localhost", 0)) .start() .block(); @@ -531,7 +531,7 @@ public class RSocketOutboundGatewayIntegrationTests { .dataMimeType("text/plain") .metadataMimeType("message/x.rsocket.routing.v0") .frameDecoder(PayloadDecoder.ZERO_COPY) - .acceptor(messageHandler().clientResponder()) + .acceptor(messageHandler().responder()) .transport(TcpClientTransport.create("localhost", server.address().getPort())) .start() .block();