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

Align RSocket module with the latest SF

Browse Source 
* 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
This commit is contained in:
Artem Bilan
2019-08-27 16:46:57 -04:00
parent 5073da8a1b
commit ff15d5265d
5 changed files with 10 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -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'

5
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java
View File

@@ -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.
* <p>
@@ -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()

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

@@ -34,12 +34,7 @@ import reactor.core.Disposable;
import reactor.core.publisher.Mono;
/**
* A client {@link AbstractRSocketConnector} extension to the RSocket server.
* <p>
* 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);

7
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java
View File

@@ -49,11 +49,6 @@ import reactor.netty.http.server.HttpServer;
/**
* A server {@link AbstractRSocketConnector} extension to accept and manage client RSocket connections.
* <p>
* 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();

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

@@ -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();