|
此版本仍在开发中,尚未被视为稳定版本。对于最新的稳定版本,请使用 Spring Integration 6.4.0! |
RSocket 支持
RSocket Spring 集成模块 (spring-integration-rsocket) 允许执行 RSocket 应用程序协议。
您需要将此依赖项包含在您的项目中:
-
Maven
-
Gradle
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-rsocket</artifactId>
<version>6.4.1-SNAPSHOT</version>
</dependency>
compile "org.springframework.integration:spring-integration-rsocket:6.4.1-SNAPSHOT"
该模块从版本 5.2 开始可用,并且基于 Spring Messaging 基础及其 RSocket 组件实现,例如RSocketRequester,RSocketMessageHandler和RSocketStrategies.
有关 RSocket 协议、术语和组件的更多信息,请参阅 Spring Framework RSocket 支持。
在通过通道适配器开始集成流处理之前,我们需要在服务器和客户端之间建立 RSocket 连接。
为此,Spring 集成 RSocket 支持提供了ServerRSocketConnector和ClientRSocketConnector的AbstractRSocketConnector.
这ServerRSocketConnector根据提供的io.rsocket.transport.ServerTransport用于接受来自客户端的连接。
内部RSocketServer实例可以使用setServerConfigurer(),以及可以配置的其他选项,例如RSocketStrategies和MimeType用于有效负载数据和标头元数据。
当setupRoute由客户端请求者提供(请参阅ClientRSocketConnector),连接的客户端将存储为RSocketRequester在 key 下由clientRSocketKeyStrategy BiFunction<Map<String, Object>, DataBuffer, Object>.
默认情况下,连接数据用于键,作为转换为具有 UTF-8 字符集的字符串的值。
这样的RSocketRequesterRegistry 可以在 Application Logic 中用于确定特定的 Client 端连接,以便与它进行交互,或将同一消息发布到所有连接的 Client 端。
从客户端建立连接时,RSocketConnectedEvent从ServerRSocketConnector.
这类似于@ConnectMapping注解。
映射模式表示接受所有客户端路由。
这*RSocketConnectedEvent可用于区分不同的路由DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER页眉。
典型的服务器配置可能如下所示:
@Bean
public RSocketStrategies rsocketStrategies() {
return RSocketStrategies.builder()
.decoder(StringDecoder.textPlainOnly())
.encoder(CharSequenceEncoder.allMimeTypes())
.dataBufferFactory(new DefaultDataBufferFactory(true))
.build();
}
@Bean
public ServerRSocketConnector serverRSocketConnector() {
ServerRSocketConnector serverRSocketConnector = new ServerRSocketConnector("localhost", 0);
serverRSocketConnector.setRSocketStrategies(rsocketStrategies());
serverRSocketConnector.setMetadataMimeType(new MimeType("message", "x.rsocket.routing.v0"));
serverRSocketConnector.setServerConfigurer((server) -> server.payloadDecoder(PayloadDecoder.ZERO_COPY));
serverRSocketConnector.setClientRSocketKeyStrategy((headers, data) -> ""
+ headers.get(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER));
return serverRSocketConnector;
}
@EventListener
public void onApplicationEvent(RSocketConnectedEvent event) {
...
}
所有选项,包括RSocketStrategiesbean 和@EventListener为RSocketConnectedEvent)是可选的。
看ServerRSocketConnectorJavaDocs 了解更多信息。
从版本 5.2.1 开始,ServerRSocketMessageHandler被提取到一个公共的顶级类中,以便与现有 RSocket 服务器建立可能的连接。
当ServerRSocketConnector随ServerRSocketMessageHandler,它不会在内部创建 RSocket 服务器,而只是将所有处理逻辑委托给提供的实例。
此外,ServerRSocketMessageHandler可以配置messageMappingCompatible标志也要处理@MessageMapping对于 RSocket 控制器,完全替换标准RSocketMessageHandler.
这在混合配置中可能很有用,当经典@MessageMapping方法与 RSocket 通道适配器一起存在于同一应用程序中,并且应用程序中存在外部配置的 RSocket 服务器。
这ClientRSocketConnector用作RSocketRequester基于RSocket通过提供的ClientTransport.
这RSocketConnector可以使用提供的RSocketConnectorConfigurer.
这setupRoute(使用可选的 templates 变量)和setupDatawith metadata 也可以在此组件上配置。
典型的客户端配置可能如下所示:
@Bean
public RSocketStrategies rsocketStrategies() {
return RSocketStrategies.builder()
.decoder(StringDecoder.textPlainOnly())
.encoder(CharSequenceEncoder.allMimeTypes())
.dataBufferFactory(new DefaultDataBufferFactory(true))
.build();
}
@Bean
public ClientRSocketConnector clientRSocketConnector() {
ClientRSocketConnector clientRSocketConnector =
new ClientRSocketConnector("localhost", serverRSocketConnector().getBoundPort().block());
clientRSocketConnector.setRSocketStrategies(rsocketStrategies());
clientRSocketConnector.setSetupRoute("clientConnect/{user}");
clientRSocketConnector.setSetupRouteVariables("myUser");
return clientRSocketConnector;
}
这些选项中的大多数(包括RSocketStrategiesbean)是可选的。
请注意我们如何在任意端口上连接到本地启动的 RSocket 服务器。
看ServerRSocketConnector.clientRSocketKeyStrategy为setupData使用案例。
另请参阅ClientRSocketConnector及其AbstractRSocketConnectorsuperclass JavaDocs 了解更多信息。
双ClientRSocketConnector和ServerRSocketConnector负责将入站通道适配器映射到其path用于路由传入 RSocket 请求的配置。
有关更多信息,请参阅下一节。
RSocket 入站网关
这RSocketInboundGateway负责接收 RSocket 请求并生成响应(如果有)。
它需要一个pathmapping 的 map,可以是类似于 MVC 请求映射或@MessageMapping语义学。
此外,(从版本 5.2.2 开始),一组交互模型(参见RSocketInteractionModel) 可以在RSocketInboundGateway按特定帧类型限制 RSocket 请求到此终端节点。
默认情况下,支持所有交互模型。
这样的豆子,根据其IntegrationRSocketEndpointimplementation (一个ReactiveMessageHandler)由ServerRSocketConnector或ClientRSocketConnector对于内部IntegrationRSocketMessageHandler用于传入请求。
一AbstractRSocketConnector可以提供给RSocketInboundGateway进行显式终端节点注册。
这样,自动检测选项就会被禁用AbstractRSocketConnector.
这RSocketStrategies也可以注射到RSocketInboundGateway或者它们是从提供的AbstractRSocketConnector覆盖任何显式注入。
解码器是从那些RSocketStrategies根据提供的requestElementType.
如果RSocketPayloadReturnValueHandler.RESPONSE_HEADER标头未在 incoming 中提供Message这RSocketInboundGateway将请求视为fireAndForgetRSocket 交互模型。
在这种情况下,RSocketInboundGateway执行普通send作导入到outputChannel.
否则,一个MonoProcessor值RSocketPayloadReturnValueHandler.RESPONSE_HEADERheader 用于向 RSocket 发送回复。
为此,一个RSocketInboundGateway执行sendAndReceiveMessageReactive作outputChannel.
这payload的消息始终为Flux根据MessagingRSocket逻辑。
当处于fireAndForgetRSocket 交互模型,消息具有 plain 转换payload.
回复payload可以是普通对象或Publisher-这RSocketInboundGateway根据RSocketStrategies.
从版本 5.3 开始,decodeFluxAsUnit选项(默认false) 已添加到RSocketInboundGateway.
默认情况下,传入Flux的转换方式是单独解码其每个事件。
这是当前@MessageMapping语义学。
恢复以前的行为或解码整个Flux作为单个单元,根据应用要求,decodeFluxAsUnit必须设置为true.
但是,目标解码逻辑依赖于Decoderselected,例如 AStringDecoder要求流中存在新的行分隔符(默认情况下)以指示字节缓冲区结束。
请参阅使用 Java 配置 RSocket 端点,以获取如何配置RSocketInboundGateway端点并处理下游的有效负载。
RSocket 出站网关
这RSocketOutboundGateway是一个AbstractReplyProducingMessageHandler在 RSocket 中执行请求,并根据 RSocket 回复(如果有)生成回复。
低级 RSocket 协议交互被委托给RSocketRequester从提供的ClientRSocketConnector或从RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER标头。
目标RSocketRequester可以从RSocketConnectedEvent或使用ServerRSocketConnector.getClientRSocketRequester()根据为连接请求映射选择的某些业务密钥的 APIServerRSocketConnector.setClientRSocketKeyStrategy().
看ServerRSocketConnectorJavaDocs 了解更多信息。
这route发送请求必须显式配置(与路径变量一起)或通过根据请求消息进行评估的 SpEL 表达式。
RSocket 交互模型可以通过RSocketInteractionModel选项或相应的表达式设置。
默认情况下,requestResponse用于常见的网关使用案例。
当请求消息负载为Publisher一个publisherElementType选项来根据RSocketStrategies在目标中提供RSocketRequester.
此选项的表达式的计算结果可以为ParameterizedTypeReference.
请参阅RSocketRequester.RequestSpec.data()JavaDocs 了解有关数据及其类型的更多信息。
RSocket 请求也可以通过metadata.
为此,一个metadataExpression针对请求消息的RSocketOutboundGateway.
这样的表达式必须计算为Map<Object, MimeType>.
什么时候interactionModel莫fireAndForget一expectedResponseType必须提供。
它是一个String.class默认情况下。
此选项的表达式的计算结果可以为ParameterizedTypeReference.
请参阅RSocketRequester.RetrieveSpec.retrieveMono()和RSocketRequester.RetrieveSpec.retrieveFlux()JavaDocs 了解有关回复数据及其类型的更多信息。
回复payload从RSocketOutboundGateway是一个Mono(即使对于fireAndForget交互模型Mono<Void>) 始终将此组件设为async.
这样的Mono在生成到outputChannel对于常规渠道,或由FluxMessageChannel.
一个Fluxresponse 的requestStream或requestChannel交互模型也被包装到一个 reply 中Mono.
它可以被FluxMessageChannel使用 PassThrough Service Activator:
@ServiceActivator(inputChannel = "rsocketReplyChannel", outputChannel ="fluxMessageChannel")
public Flux<?> flattenRSocketResponse(Flux<?> payload) {
return payload;
}
或在目标应用程序逻辑中显式订阅。
预期的响应类型也可以配置(或通过表达式评估)为void将此网关视为出站通道适配器。
但是,outputChannel仍然必须配置(即使它只是一个NullChannel) 启动对返回的Mono.
请参阅使用 Java 配置 RSocket 端点,以获取如何配置RSocketOutboundGateway端点 A 处理下游有效负载。
RSocket 命名空间支持
Spring 集成提供了一个rsocketnamespace 和相应的 schema definition 创建。
要将其包含在您的配置中,请在您的应用程序上下文配置文件中添加以下命名空间声明:
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-rsocket="http://www.springframework.org/schema/integration/rsocket"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/rsocket
https://www.springframework.org/schema/integration/rsocket/spring-integration-rsocket.xsd">
...
</beans>
入境
要使用 XML 配置 Spring 集成 RSocket 入站通道适配器,您需要使用适当的inbound-gateway组件int-rsocketNamespace。
以下示例显示如何配置它:
<int-rsocket:inbound-gateway id="inboundGateway"
path="testPath"
interaction-models="requestStream,requestChannel"
rsocket-connector="clientRSocketConnector"
request-channel="requestChannel"
rsocket-strategies="rsocketStrategies"
request-element-type="byte[]"/>
一个ClientRSocketConnector和ServerRSocketConnector应配置为通用<bean>定义。
出境
<int-rsocket:outbound-gateway id="outboundGateway"
client-rsocket-connector="clientRSocketConnector"
auto-startup="false"
interaction-model="fireAndForget"
route-expression="'testRoute'"
request-channel="requestChannel"
publisher-element-type="byte[]"
expected-response-type="java.util.Date"
metadata-expression="{'metadata': new org.springframework.util.MimeType('*')}"/>
看spring-integration-rsocket.xsd以获取所有这些 XML 属性的描述。
使用 Java 配置 RSocket 端点
以下示例显示了如何使用 Java 配置 RSocket 入站端点:
@Bean
public RSocketInboundGateway rsocketInboundGatewayRequestReply() {
RSocketInboundGateway rsocketInboundGateway = new RSocketInboundGateway("echo");
rsocketInboundGateway.setRequestChannelName("requestReplyChannel");
return rsocketInboundGateway;
}
@Transformer(inputChannel = "requestReplyChannel")
public Mono<String> echoTransformation(Flux<String> payload) {
return payload.next().map(String::toUpperCase);
}
一个ClientRSocketConnector或ServerRSocketConnector在此配置中假定,其含义是在 “echo” 路径上自动检测此类端点。
注意@Transformer签名,它对 RSocket 请求的完全反应式处理并生成反应式回复。
以下示例显示了如何使用 Java DSL 配置 RSocket 入站网关:
@Bean
public IntegrationFlow rsocketUpperCaseFlow() {
return IntegrationFlow
.from(RSockets.inboundGateway("/uppercase")
.interactionModels(RSocketInteractionModel.requestChannel))
.<Flux<String>, Mono<String>>transform((flux) -> flux.next().map(String::toUpperCase))
.get();
}
一个ClientRSocketConnector或ServerRSocketConnector在此配置中假定其含义是在 “/uppercase” 路径上自动检测此类端点,并将预期的交互模型视为 “request channel”。
以下示例显示如何使用 Java 配置 RSocket 出站网关:
@Bean
@ServiceActivator(inputChannel = "requestChannel", outputChannel = "replyChannel")
public RSocketOutboundGateway rsocketOutboundGateway() {
RSocketOutboundGateway rsocketOutboundGateway =
new RSocketOutboundGateway(
new FunctionExpression<Message<?>>((m) ->
m.getHeaders().get("route_header")));
rsocketOutboundGateway.setInteractionModelExpression(
new FunctionExpression<Message<?>>((m) -> m.getHeaders().get("rsocket_interaction_model")));
rsocketOutboundGateway.setClientRSocketConnector(clientRSocketConnector());
return rsocketOutboundGateway;
}
这setClientRSocketConnector()仅对客户端是必需的。
在服务器端,RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER标头中带有RSocketRequester值。
以下示例显示了如何使用 Java DSL 配置 RSocket 出站网关:
@Bean
public IntegrationFlow rsocketUpperCaseRequestFlow(ClientRSocketConnector clientRSocketConnector) {
return IntegrationFlow
.from(Function.class)
.handle(RSockets.outboundGateway("/uppercase")
.interactionModel(RSocketInteractionModel.requestResponse)
.expectedResponseType(String.class)
.clientRSocketConnector(clientRSocketConnector))
.get();
}
看IntegrationFlow作为网关有关更多信息,请参阅如何使用上述Function接口。