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

Document RSocket support

Browse Source 
* Fix some typos in `webflux.adoc` and `dsl.adoc`
* Make `FunctionsTests.kt` more Kotlin friendly
* Improve RSocket components and tests for them,
especially rely on the default `RSocketStrategies` in the
target SF RSocket components
* Rework `ServerRSocketConnector.setClientRSocketKeyStrategy` to use the whole `MessageHeaders` for consultation.
It turns out that just destination is not enough since it can be used from different clients
* Make the key based on the provided `setupData` by default
* Include all the `MessageHeaders` into the `RSocketConnectedEvent`

* Improve Docs according review feedback and Docs in SF

Doc polishing
This commit is contained in:
Artem Bilan
2019-09-20 09:24:26 -04:00
committed by Gary Russell
parent 21335812a3
commit b12657c599
12 changed files with 380 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -23,10 +23,6 @@ import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.codec.CharSequenceEncoder;
import org.springframework.core.codec.StringDecoder;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.messaging.rsocket.MetadataExtractor;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
@@ -38,7 +34,7 @@ import io.rsocket.metadata.WellKnownMimeType;
* A base connector container for common RSocket client and server functionality.
* <p>
* It accepts {@link IntegrationRSocketEndpoint} instances for mapping registration via an internal
* {@link IntegrationRSocketMessageHandler} or performs an auto-detection otherwise, when all bean are ready
* {@link IntegrationRSocketMessageHandler} or performs an auto-detection otherwise, when all beans are ready
* in the application context.
*
* @author Artem Bilan
@@ -58,12 +54,7 @@ public abstract class AbstractRSocketConnector
private MimeType metadataMimeType =
MimeTypeUtils.parseMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.toString());
private RSocketStrategies rsocketStrategies =
RSocketStrategies.builder()
.decoder(StringDecoder.allMimeTypes())
.encoder(CharSequenceEncoder.allMimeTypes())
.dataBufferFactory(new DefaultDataBufferFactory())
.build();
private RSocketStrategies rsocketStrategies = RSocketStrategies.create();
private boolean autoStartup = true;
@@ -114,7 +105,7 @@ public abstract class AbstractRSocketConnector
}
/**
* Configure {@link IntegrationRSocketEndpoint} instances for mapping nad handling requests.
* Configure {@link IntegrationRSocketEndpoint} instances for mapping and handling requests.
* @param endpoints the {@link IntegrationRSocketEndpoint} instances for handling inbound requests.
* @see #addEndpoint(IntegrationRSocketEndpoint)
*/
@@ -125,20 +116,6 @@ public abstract class AbstractRSocketConnector
}
}
/**
* Configure a {@link MetadataExtractor} to extract the route and possibly
* other metadata from the first payload of incoming requests.
* <p>By default this is a
* {@link org.springframework.messaging.rsocket.DefaultMetadataExtractor}
* with the configured {@link RSocketStrategies} (and decoders), extracting a route
* from {@code "message/x.rsocket.routing.v0"} or {@code "text/plain"}
* metadata entries.
* @param extractor the extractor to use
*/
public void setMetadataExtractor(MetadataExtractor extractor) {
this.rSocketMessageHandler.setMetadataExtractor(extractor);
}
/**
* Add an {@link IntegrationRSocketEndpoint} for mapping and handling RSocket requests.
* @param endpoint the {@link IntegrationRSocketEndpoint} to map.

19
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/RSocketConnectedEvent.java
View File

@@ -16,6 +16,8 @@
package org.springframework.integration.rsocket;
import java.util.Map;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.integration.events.IntegrationEvent;
import org.springframework.messaging.rsocket.RSocketRequester;
@@ -25,7 +27,7 @@ import org.springframework.messaging.rsocket.RSocketRequester;
* to the server.
* <p>
* This event can be used for mapping {@link RSocketRequester} to the client by the
* {@code destination} meta-data or connect payload {@code data}.
* {@code headers} meta-data or connect payload {@code data}.
*
* @author Artem Bilan
*
@@ -36,21 +38,23 @@ import org.springframework.messaging.rsocket.RSocketRequester;
@SuppressWarnings("serial")
public class RSocketConnectedEvent extends IntegrationEvent {
private final String destination;
private final Map<String, Object> headers;
private final DataBuffer data;
private final RSocketRequester requester;
public RSocketConnectedEvent(Object source, String destination, DataBuffer data, RSocketRequester requester) {
public RSocketConnectedEvent(Object source, Map<String, Object> headers, DataBuffer data,
RSocketRequester requester) {
super(source);
this.destination = destination;
this.headers = headers;
this.data = data;
this.requester = requester;
}
public String getDestination() {
return this.destination;
public Map<String, Object> getHeaders() {
return this.headers;
}
public DataBuffer getData() {
@@ -64,7 +68,8 @@ public class RSocketConnectedEvent extends IntegrationEvent {
@Override
public String toString() {
return "RSocketConnectedEvent{" +
"destination='" + this.destination + '\'' +
"headers=" + this.headers +
", data=" + this.data +
", requester=" + this.requester +
'}';
}

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

@@ -17,6 +17,7 @@
package org.springframework.integration.rsocket;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -36,7 +37,6 @@ import org.springframework.messaging.rsocket.annotation.support.RSocketFrameType
import org.springframework.messaging.rsocket.annotation.support.RSocketRequesterMethodArgumentResolver;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.RouteMatcher;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.ServerTransport;
@@ -108,7 +108,9 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
* Defaults to the {@code destination} the client is connected.
* @param clientRSocketKeyStrategy the {@link BiFunction} to determine a key for client {@link RSocketRequester}s.
*/
public void setClientRSocketKeyStrategy(BiFunction<String, DataBuffer, Object> clientRSocketKeyStrategy) {
public void setClientRSocketKeyStrategy(BiFunction<Map<String, Object>,
DataBuffer, Object> clientRSocketKeyStrategy) {
Assert.notNull(clientRSocketKeyStrategy, "'clientRSocketKeyStrategy' must not be null");
serverRSocketMessageHandler().clientRSocketKeyStrategy = clientRSocketKeyStrategy;
}
@@ -176,7 +178,8 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
private final Map<Object, RSocketRequester> clientRSocketRequesters = new HashMap<>();
private BiFunction<String, DataBuffer, Object> clientRSocketKeyStrategy = (destination, data) -> destination;
private BiFunction<Map<String, Object>, DataBuffer, Object> clientRSocketKeyStrategy =
(headers, data) -> data.toString(StandardCharsets.UTF_8);
private ApplicationEventPublisher applicationEventPublisher;
@@ -184,28 +187,20 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
registerHandlerMethod(this, HANDLE_CONNECTION_SETUP_METHOD,
new CompositeMessageCondition(
RSocketFrameTypeMessageCondition.CONNECT_CONDITION,
new DestinationPatternsMessageCondition(new String[] { "*" }, getRouteMatcher()))); // NOSONAR
new DestinationPatternsMessageCondition(new String[] { "*" }, obtainRouteMatcher())));
}
@SuppressWarnings("unused")
private void handleConnectionSetup(Message<DataBuffer> connectMessage) {
DataBuffer dataBuffer = connectMessage.getPayload();
MessageHeaders messageHeaders = connectMessage.getHeaders();
String destination = "";
RouteMatcher.Route route =
messageHeaders.get(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER,
RouteMatcher.Route.class);
if (route != null) {
destination = route.value();
}
Object rsocketRequesterKey = this.clientRSocketKeyStrategy.apply(destination, dataBuffer);
Object rsocketRequesterKey = this.clientRSocketKeyStrategy.apply(messageHeaders, dataBuffer);
RSocketRequester rsocketRequester =
messageHeaders.get(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER,
RSocketRequester.class);
this.clientRSocketRequesters.put(rsocketRequesterKey, rsocketRequester);
RSocketConnectedEvent rSocketConnectedEvent =
new RSocketConnectedEvent(this, destination, dataBuffer, rsocketRequester); // NOSONAR
new RSocketConnectedEvent(this, messageHeaders, dataBuffer, rsocketRequester); // NOSONAR
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(rSocketConnectedEvent);
}

12
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java
View File

@@ -22,13 +22,10 @@ import org.reactivestreams.Publisher;
import org.springframework.core.ReactiveAdapter;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.CharSequenceEncoder;
import org.springframework.core.codec.Decoder;
import org.springframework.core.codec.Encoder;
import org.springframework.core.codec.StringDecoder;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.rsocket.AbstractRSocketConnector;
import org.springframework.integration.rsocket.ClientRSocketConnector;
@@ -77,12 +74,7 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In
private final String[] path;
private RSocketStrategies rsocketStrategies =
RSocketStrategies.builder()
.decoder(StringDecoder.allMimeTypes())
.encoder(CharSequenceEncoder.allMimeTypes())
.dataBufferFactory(new DefaultDataBufferFactory())
.build();
private RSocketStrategies rsocketStrategies = RSocketStrategies.create();
@Nullable
private AbstractRSocketConnector rsocketConnector;
@@ -91,7 +83,7 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In
private ResolvableType requestElementType;
/**
* Instantiate based on the provided Ant-style path patterns to map this endpoint for incoming RSocket requests.
* Instantiate based on the provided path patterns to map this endpoint for incoming RSocket requests.
* @param pathArg the mapping patterns to use.
*/
public RSocketInboundGateway(String... pathArg) {

25
spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java
View File

@@ -26,21 +26,15 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.codec.CharSequenceEncoder;
import org.springframework.core.codec.StringDecoder;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.rsocket.ClientRSocketConnector;
import org.springframework.integration.rsocket.ServerRSocketConnector;
import org.springframework.integration.rsocket.outbound.RSocketOutboundGateway;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import io.netty.buffer.PooledByteBufAllocator;
import io.rsocket.frame.decoder.PayloadDecoder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -66,29 +60,15 @@ public class RSocketDslTests {
@EnableIntegration
public static class TestConfiguration {
@Bean
public RSocketStrategies rsocketStrategies() {
return RSocketStrategies.builder()
.decoder(StringDecoder.allMimeTypes())
.encoder(CharSequenceEncoder.allMimeTypes())
.dataBufferFactory(new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT))
.build();
}
@Bean
public ServerRSocketConnector serverRSocketConnector() {
ServerRSocketConnector serverRSocketConnector = new ServerRSocketConnector("localhost", 0);
serverRSocketConnector.setRSocketStrategies(rsocketStrategies());
serverRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY));
return serverRSocketConnector;
return new ServerRSocketConnector("localhost", 0);
}
@Bean
public ClientRSocketConnector clientRSocketConnector(ServerRSocketConnector serverRSocketConnector) {
int port = serverRSocketConnector.getBoundPort().block();
ClientRSocketConnector clientRSocketConnector = new ClientRSocketConnector("localhost", port);
clientRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY));
clientRSocketConnector.setRSocketStrategies(rsocketStrategies());
clientRSocketConnector.setAutoStartup(false);
return clientRSocketConnector;
}
@@ -107,8 +87,7 @@ public class RSocketDslTests {
@Bean
public IntegrationFlow rsocketUpperCaseFlow() {
return IntegrationFlows
.from(RSockets.inboundGateway("/uppercase")
.rsocketStrategies(rsocketStrategies()))
.from(RSockets.inboundGateway("/uppercase"))
.<Flux<String>, Mono<String>>transform((flux) -> flux.next().map(String::toUpperCase))
.get();
}

52
spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/inbound/RSocketInboundGatewayIntegrationTests.java
View File

@@ -27,15 +27,11 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.codec.CharSequenceEncoder;
import org.springframework.core.codec.StringDecoder;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.context.event.EventListener;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.rsocket.ClientRSocketConnector;
@@ -44,13 +40,9 @@ import org.springframework.integration.rsocket.ServerRSocketConnector;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.util.MimeType;
import io.netty.buffer.PooledByteBufAllocator;
import io.rsocket.frame.decoder.PayloadDecoder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
@@ -154,15 +146,6 @@ public class RSocketInboundGatewayIntegrationTests {
private abstract static class CommonConfig {
@Bean
public RSocketStrategies rsocketStrategies() {
return RSocketStrategies.builder()
.decoder(StringDecoder.allMimeTypes())
.encoder(CharSequenceEncoder.allMimeTypes())
.dataBufferFactory(new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT))
.build();
}
@Bean
public PollableChannel fireAndForgetChannelChannel() {
return new QueueChannel();
@@ -171,7 +154,6 @@ public class RSocketInboundGatewayIntegrationTests {
@Bean
public RSocketInboundGateway rsocketInboundGatewayFireAndForget() {
RSocketInboundGateway rsocketInboundGateway = new RSocketInboundGateway("receive");
rsocketInboundGateway.setRSocketStrategies(rsocketStrategies());
rsocketInboundGateway.setRequestChannel(fireAndForgetChannelChannel());
return rsocketInboundGateway;
}
@@ -179,16 +161,10 @@ public class RSocketInboundGatewayIntegrationTests {
@Bean
public RSocketInboundGateway rsocketInboundGatewayRequestReply() {
RSocketInboundGateway rsocketInboundGateway = new RSocketInboundGateway("echo");
rsocketInboundGateway.setRSocketStrategies(rsocketStrategies());
rsocketInboundGateway.setRequestChannel(requestReplyChannel());
rsocketInboundGateway.setRequestChannelName("requestReplyChannel");
return rsocketInboundGateway;
}
@Bean
public FluxMessageChannel requestReplyChannel() {
return new FluxMessageChannel();
}
@Transformer(inputChannel = "requestReplyChannel")
public Mono<String> echoTransformation(Flux<String> payload) {
return payload.next().map(String::toUpperCase);
@@ -198,22 +174,18 @@ public class RSocketInboundGatewayIntegrationTests {
@Configuration
@EnableIntegration
static class ServerConfig extends CommonConfig implements ApplicationListener<RSocketConnectedEvent> {
static class ServerConfig extends CommonConfig {
final MonoProcessor<RSocketRequester> clientRequester = MonoProcessor.create();
@Override
public void onApplicationEvent(RSocketConnectedEvent event) {
this.clientRequester.onNext(event.getRequester());
}
@Bean
public ServerRSocketConnector serverRSocketConnector() {
ServerRSocketConnector serverRSocketConnector = new ServerRSocketConnector("localhost", 0);
serverRSocketConnector.setRSocketStrategies(rsocketStrategies());
serverRSocketConnector.setMetadataMimeType(new MimeType("message", "x.rsocket.routing.v0"));
serverRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY));
return serverRSocketConnector;
return new ServerRSocketConnector("localhost", 0);
}
@EventListener
public void onApplicationEvent(RSocketConnectedEvent event) {
this.clientRequester.onNext(event.getRequester());
}
}
@@ -227,10 +199,8 @@ public class RSocketInboundGatewayIntegrationTests {
ClientRSocketConnector clientRSocketConnector =
new ClientRSocketConnector("localhost",
serverConfig.serverRSocketConnector().getBoundPort().block());
clientRSocketConnector.setMetadataMimeType(new MimeType("message", "x.rsocket.routing.v0"));
clientRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY));
clientRSocketConnector.setRSocketStrategies(rsocketStrategies());
clientRSocketConnector.setSetupRoute("clientConnect");
clientRSocketConnector.setSetupRoute("clientConnect/{user}");
clientRSocketConnector.setSetupRouteVariables("myUser");
return clientRSocketConnector;
}

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

@@ -30,9 +30,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.codec.CharSequenceEncoder;
import org.springframework.core.codec.StringDecoder;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
@@ -59,7 +56,6 @@ import org.springframework.stereotype.Controller;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import io.netty.buffer.PooledByteBufAllocator;
import io.rsocket.RSocket;
import io.rsocket.RSocketFactory;
import io.rsocket.frame.decoder.PayloadDecoder;
@@ -468,15 +464,6 @@ public class RSocketOutboundGatewayIntegrationTests {
private abstract static class CommonConfig {
@Bean
public RSocketStrategies rsocketStrategies() {
return RSocketStrategies.builder()
.decoder(StringDecoder.allMimeTypes())
.encoder(CharSequenceEncoder.allMimeTypes())
.dataBufferFactory(new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT))
.build();
}
@Bean
public TestController controller() {
return new TestController();
@@ -515,10 +502,9 @@ public class RSocketOutboundGatewayIntegrationTests {
@Bean(destroyMethod = "dispose")
@Nullable
public RSocket rsocketForServerRequests() {
return RSocketRequester.builder()
.setupRoute("clientConnect")
.rsocketFactory(RSocketMessageHandler.clientResponder(rsocketStrategies(), controller()))
.rsocketFactory(RSocketMessageHandler.clientResponder(RSocketStrategies.create(), controller()))
.connectTcp("localhost", server.address().getPort())
.block()
.rsocket();
@@ -526,10 +512,7 @@ public class RSocketOutboundGatewayIntegrationTests {
@Bean
public ClientRSocketConnector clientRSocketConnector() {
ClientRSocketConnector clientRSocketConnector =
new ClientRSocketConnector("localhost", server.address().getPort());
clientRSocketConnector.setRSocketStrategies(rsocketStrategies());
return clientRSocketConnector;
return new ClientRSocketConnector("localhost", server.address().getPort());
}
@Override
@@ -548,9 +531,7 @@ public class RSocketOutboundGatewayIntegrationTests {
@Bean
public RSocketMessageHandler messageHandler() {
RSocketMessageHandler handler = new RSocketMessageHandler();
handler.setRSocketStrategies(rsocketStrategies());
return handler;
return new RSocketMessageHandler();
}
}