From 89e11f2c464bddb689ba912c20e79fcecd556603 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 25 Apr 2019 14:07:56 -0400 Subject: [PATCH] Add initial support for RSockets (#2902) * Add initial support for RSockets * Add `spring-integration-rsocket` module and respective dependencies * Implement `RSocketOutboundGateway` based on the Spring Messaging `RSocketRequester`. This component supports dynamic RSocket properties via expressions against request message. to handle `Publisher` for requests, it must be present in the request message `payload` instead of `FluxMessageChannel` upstream, since the last one just flattens events to be handled in the `MessageHandler` one by one. The result `Mono` is subscribed downstream in the `FluxMessageChannel` or directly by the `AbstractReplyProducingMessageHandler`. If result is a `Flux` it is just wrapped into the `Mono` to be processed downstream by end-user code. The point is that these request/replies are volatile and live in the particular context meanwhile a `FluxMessageChannel` is long living publisher in the application context boundaries. * The `RSocketOutboundGatewayIntegrationTests` is an adapted copy of `RSocketClientToServerIntegrationTests` from Spring Messaging * Add `doOnError()` into the `Flux` created in the `AbstractMessageProducingHandler` for `Publisher` replies * * Use singular for the `RSocket` term * Use no-op `Consumer` for the `strategiesConfigurer` and `factoryConfigurer` in the `RSocketOutboundGateway` and also `Assert.notNull()` in the appropriate setters to avoid null check during `RSocketRequester.builder()` initialization * Use `TcpServer.create().port(0)` in the `RSocketOutboundGatewayIntegrationTests` to allow to select free OS port and bind into it. The selected port is used later for client configuration in the `RSocketOutboundGateway` bean definition * * Change `RSocketOutboundGatewayIntegrationTests.PORT` to lower case --- build.gradle | 20 +- .../channel/FluxMessageChannel.java | 4 +- .../AbstractMessageProducingHandler.java | 13 +- .../AbstractReplyProducingMessageHandler.java | 3 + .../config/RSocketNamespaceHandler.java | 32 ++ .../outbound/RSocketOutboundGateway.java | 299 ++++++++++++ .../rsocket/outbound/package-info.java | 4 + .../main/resources/META-INF/spring.handlers | 1 + .../main/resources/META-INF/spring.schemas | 2 + .../main/resources/META-INF/spring.tooling | 4 + .../config/spring-integration-rsocket-5.2.xsd | 21 + .../config/spring-integration-rsocket.gif | Bin 0 -> 578 bytes ...SocketOutboundGatewayIntegrationTests.java | 439 ++++++++++++++++++ .../src/test/resources/log4j2-test.xml | 16 + 14 files changed, 846 insertions(+), 12 deletions(-) create mode 100644 spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketNamespaceHandler.java create mode 100644 spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java create mode 100644 spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/package-info.java create mode 100644 spring-integration-rsocket/src/main/resources/META-INF/spring.handlers create mode 100644 spring-integration-rsocket/src/main/resources/META-INF/spring.schemas create mode 100644 spring-integration-rsocket/src/main/resources/META-INF/spring.tooling create mode 100644 spring-integration-rsocket/src/main/resources/org/springframework/integration/rsocket/config/spring-integration-rsocket-5.2.xsd create mode 100644 spring-integration-rsocket/src/main/resources/org/springframework/integration/rsocket/config/spring-integration-rsocket.gif create mode 100644 spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java create mode 100644 spring-integration-rsocket/src/test/resources/log4j2-test.xml diff --git a/build.gradle b/build.gradle index 259f8439de..1c5cbc6532 100644 --- a/build.gradle +++ b/build.gradle @@ -39,6 +39,7 @@ allprojects { if (version.endsWith('BUILD-SNAPSHOT')) { maven { url 'https://repo.spring.io/libs-snapshot' } } + maven { url "https://oss.jfrog.org/artifactory/libs-snapshot" } // RSocket // maven { url 'https://repo.spring.io/libs-staging-local' } } @@ -128,10 +129,11 @@ subprojects { subproject -> mysqlVersion = '8.0.15' pahoMqttClientVersion = '1.2.0' postgresVersion = '42.2.5' - reactorNettyVersion = '0.8.6.RELEASE' - reactorVersion = '3.2.8.RELEASE' + reactorNettyVersion = '0.9.0.BUILD-SNAPSHOT' + reactorVersion = '3.3.0.BUILD-SNAPSHOT' resilience4jVersion = '0.14.1' romeToolsVersion = '1.12.0' + rsocketVersion = '0.12.2-RC3-SNAPSHOT' servletApiVersion = '4.0.1' smackVersion = '4.3.3' springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.2.0.M1' @@ -141,7 +143,7 @@ subprojects { subproject -> springGemfireVersion = '2.2.0.M3' springSecurityVersion = '5.2.0.M2' springRetryVersion = '1.2.4.RELEASE' - springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.2.0.M1' + springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.2.0.BUILD-SNAPSHOT' springWsVersion = '3.0.7.RELEASE' tomcatVersion = "9.0.17" xstreamVersion = '1.4.11.1' @@ -596,6 +598,18 @@ project('spring-integration-rmi') { } } +project('spring-integration-rsocket') { + description = 'Spring Integration RSocket Support' + dependencies { + compile project(":spring-integration-core") + compile("io.projectreactor.netty:reactor-netty:$reactorNettyVersion") + compile("io.rsocket:rsocket-core:$rsocketVersion") + compile("io.rsocket:rsocket-transport-netty:$rsocketVersion") + + testCompile "io.projectreactor:reactor-test:$reactorVersion" + } +} + project('spring-integration-scripting') { description = 'Spring Integration Scripting Support' dependencies { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java index 28439164fa..ca2075a2bc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java @@ -82,8 +82,8 @@ public class FluxMessageChannel extends AbstractMessageChannel ConnectableFlux connectableFlux = Flux.from(publisher) .handle((message, sink) -> sink.next(send(message))) - .onErrorContinue((throwable, o) -> logger.warn("Error during processing event: " + o, throwable) - ) + .onErrorContinue((throwable, event) -> + logger.warn("Error during processing event: " + event, throwable)) .doOnComplete(() -> this.publishers.remove(publisher)) .publish(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index 1060335ecd..60611633f0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -29,6 +29,7 @@ import org.reactivestreams.Publisher; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel; +import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.routingslip.RoutingSlipRouteStrategy; @@ -283,6 +284,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan ((ReactiveStreamsSubscribableChannel) messageChannel) .subscribeTo( Flux.from((Publisher) reply) + .doOnError((ex) -> sendErrorMessage(requestMessage, ex)) .map(result -> createOutputMessage(result, requestHeaders))); } } @@ -311,25 +313,22 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan } private void asyncNonReactiveReply(Message requestMessage, Object reply, Object replyChannel) { - ListenableFuture future; if (reply instanceof ListenableFuture) { future = (ListenableFuture) reply; } else { SettableListenableFuture settableListenableFuture = new SettableListenableFuture<>(); - Mono.from((Publisher) reply) .subscribe(settableListenableFuture::set, settableListenableFuture::setException); - future = settableListenableFuture; } - future.addCallback(new ReplyFutureCallback(requestMessage, replyChannel)); } private Object getOutputChannelFromRoutingSlip(Object reply, Message requestMessage, List routingSlip, AtomicInteger routingSlipIndex) { + if (routingSlipIndex.get() >= routingSlip.size()) { return null; } @@ -365,7 +364,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan } protected Message createOutputMessage(Object output, MessageHeaders requestHeaders) { - AbstractIntegrationMessageBuilder builder = null; + AbstractIntegrationMessageBuilder builder; if (output instanceof Message) { if (this.noHeadersPropagation || !shouldCopyRequestHeaders()) { return (Message) output; @@ -449,7 +448,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan } catch (Exception e) { Exception exceptionToLog = - IntegrationUtils.wrapInHandlingExceptionIfNecessary(requestMessage, () -> null, e); + IntegrationUtils.wrapInHandlingExceptionIfNecessary(requestMessage, () -> null, e); logger.error("Failed to send async reply", exceptionToLog); } } @@ -459,7 +458,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan Object errorChannel = requestHeaders.getErrorChannel(); if (errorChannel == null) { try { - errorChannel = getChannelResolver().resolveDestination("errorChannel"); + errorChannel = getChannelResolver().resolveDestination(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME); } catch (DestinationResolutionException e) { // ignore diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index 997dc3e991..7be16dc673 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java @@ -87,6 +87,9 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa this.beanClassLoader = beanClassLoader; } + protected ClassLoader getBeanClassLoader() { + return this.beanClassLoader; + } @Override protected final void onInit() { diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketNamespaceHandler.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketNamespaceHandler.java new file mode 100644 index 0000000000..e12159febf --- /dev/null +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketNamespaceHandler.java @@ -0,0 +1,32 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.rsocket.config; + +import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler; + +/** + * Namespace handler for Spring Integration's RSocket namespace. + * + * @author Artem Bilan + */ +public class RSocketNamespaceHandler extends AbstractIntegrationNamespaceHandler { + + public void init() { + + } + +} diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java new file mode 100644 index 0000000000..95ffbf0156 --- /dev/null +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java @@ -0,0 +1,299 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.rsocket.outbound; + +import java.util.function.Consumer; + +import org.reactivestreams.Publisher; + +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.integration.expression.ExpressionUtils; +import org.springframework.integration.expression.ValueExpression; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.messaging.Message; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketStrategies; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; +import org.springframework.util.MimeType; +import org.springframework.util.MimeTypeUtils; + +import io.rsocket.RSocket; +import io.rsocket.RSocketFactory; +import io.rsocket.transport.ClientTransport; +import reactor.core.publisher.Mono; + +/** + * An Outbound Messaging Gateway for RSocket client requests. + * + * @author Artem Bilan + * + * @since 5.2 + * + * @see RSocketRequester + */ +public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler { + + private final ClientTransport clientTransport; + + private final Expression routeExpression; + + private MimeType dataMimeType = MimeTypeUtils.TEXT_PLAIN; + + private Consumer factoryConfigurer = (clientRSocketFactory) -> { }; + + private Consumer strategiesConfigurer = (builder) -> { }; + + private Expression commandExpression = new ValueExpression<>(Command.requestResponse); + + private Expression publisherElementTypeExpression = new ValueExpression<>(String.class); + + private Expression expectedResponseTypeExpression = new ValueExpression<>(String.class); + + private Mono rSocketRequesterMono; + + private EvaluationContext evaluationContext; + + public RSocketOutboundGateway(ClientTransport clientTransport, String route) { + this(clientTransport, new ValueExpression<>(route)); + } + + public RSocketOutboundGateway(ClientTransport clientTransport, Expression routeExpression) { + Assert.notNull(clientTransport, "'clientTransport' must not be null"); + Assert.notNull(routeExpression, "'routeExpression' must not be null"); + this.clientTransport = clientTransport; + this.routeExpression = routeExpression; + setAsync(true); + setPrimaryExpression(this.routeExpression); + } + + public void setDataMimeType(MimeType dataMimeType) { + Assert.notNull(dataMimeType, "'dataMimeType' must not be null"); + this.dataMimeType = dataMimeType; + } + + public void setFactoryConfigurer(Consumer factoryConfigurer) { + Assert.notNull(factoryConfigurer, "'factoryConfigurer' must not be null"); + this.factoryConfigurer = factoryConfigurer; + } + + public void setStrategiesConfigurer(Consumer strategiesConfigurer) { + Assert.notNull(strategiesConfigurer, "'strategiesConfigurer' must not be null"); + this.strategiesConfigurer = strategiesConfigurer; + } + + public void setCommand(Command command) { + setCommandExpression(new ValueExpression<>(command)); + } + + public void setCommandExpression(Expression commandExpression) { + Assert.notNull(commandExpression, "'commandExpression' must not be null"); + this.commandExpression = commandExpression; + } + + /** + * Configure a type for a request {@link Publisher} elements. + * @param publisherElementType the type of the request {@link Publisher} elements. + * @see RSocketRequester.RequestSpec#data(Publisher, Class) + */ + public void setPublisherElementType(Class publisherElementType) { + Assert.notNull(publisherElementType, "'publisherElementType' must not be null"); + setPublisherElementTypeExpression(new ValueExpression<>(publisherElementType)); + + } + + /** + * Configure a SpEL expression to evaluate a request {@link Publisher} elements type at runtime against + * a request message. + * @param publisherElementTypeExpression the expression to evaluate a type for the request + * {@link Publisher} elements. + * @see RSocketRequester.RequestSpec#data + */ + public void setPublisherElementTypeExpression(Expression publisherElementTypeExpression) { + this.publisherElementTypeExpression = publisherElementTypeExpression; + } + + /** + * Specify the expected response type for the RSocket response. + * @param expectedResponseType The expected type. + * @see #setExpectedResponseTypeExpression(Expression) + * @see RSocketRequester.ResponseSpec#retrieveMono + * @see RSocketRequester.ResponseSpec#retrieveFlux + */ + public void setExpectedResponseType(Class expectedResponseType) { + Assert.notNull(expectedResponseType, "'expectedResponseType' must not be null"); + setExpectedResponseTypeExpression(new ValueExpression<>(expectedResponseType)); + } + + /** + * Specify the {@link Expression} to determine the type for the RSocket response. + * @param expectedResponseTypeExpression The expected response type expression. + * @see RSocketRequester.ResponseSpec#retrieveMono + * @see RSocketRequester.ResponseSpec#retrieveFlux + */ + public void setExpectedResponseTypeExpression(Expression expectedResponseTypeExpression) { + this.expectedResponseTypeExpression = expectedResponseTypeExpression; + } + + + @Override + protected void doInit() { + super.doInit(); + this.rSocketRequesterMono = + RSocketRequester.builder() + .rsocketFactory(this.factoryConfigurer) + .rsocketStrategies(this.strategiesConfigurer) + .connect(this.clientTransport, this.dataMimeType); + + this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); + } + + @Override + public void destroy() { + super.destroy(); + this.rSocketRequesterMono.block().rsocket().dispose(); + } + + @Override + protected Object handleRequestMessage(Message requestMessage) { + return this.rSocketRequesterMono.cache() + .map((rSocketRequester) -> createRequestSpec(rSocketRequester, requestMessage)) + .map((requestSpec) -> createResponseSpec(requestSpec, requestMessage)) + .flatMap((responseSpec) -> performRequest(responseSpec, requestMessage)); + } + + private RSocketRequester.RequestSpec createRequestSpec(RSocketRequester rSocketRequester, + Message requestMessage) { + + String route = this.routeExpression.getValue(this.evaluationContext, requestMessage, String.class); + Assert.notNull(route, () -> "The 'routeExpression' [" + this.routeExpression + "] must not evaluate to null"); + + return rSocketRequester.route(route); + } + + private RSocketRequester.ResponseSpec createResponseSpec(RSocketRequester.RequestSpec requestSpec, + Message requestMessage) { + + Object payload = requestMessage.getPayload(); + + if (payload instanceof Publisher) { + Object publisherElementType = evaluateExpressionForType(requestMessage, this.publisherElementTypeExpression, + "publisherElementTypeExpression"); + return responseSpecForPublisher(requestSpec, (Publisher) payload, publisherElementType); + } + else { + return requestSpec.data(payload); + } + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + private RSocketRequester.ResponseSpec responseSpecForPublisher(RSocketRequester.RequestSpec requestSpec, + Publisher payload, Object publisherElementType) { + + if (publisherElementType instanceof Class) { + return requestSpec.data(payload, (Class) publisherElementType); + } + else { + return requestSpec.data(payload, (ParameterizedTypeReference) publisherElementType); + } + } + + private Mono performRequest(RSocketRequester.ResponseSpec responseSpec, Message requestMessage) { + Command command = this.commandExpression.getValue(this.evaluationContext, requestMessage, Command.class); + Assert.notNull(command, () -> "The 'commandExpression' [" + this.commandExpression + + "] must not evaluate to null"); + + Object expectedResponseType = null; + if (!Command.fireAndForget.equals(command)) { + expectedResponseType = evaluateExpressionForType(requestMessage, this.expectedResponseTypeExpression, + "expectedResponseTypeExpression"); + } + + switch (command) { + case fireAndForget: + return responseSpec.send(); + case requestResponse: + if (expectedResponseType instanceof Class) { + return responseSpec.retrieveMono((Class) expectedResponseType); + } + else { + return responseSpec.retrieveMono((ParameterizedTypeReference) expectedResponseType); + } + case requestStreamOrChannel: + if (expectedResponseType instanceof Class) { + return Mono.just(responseSpec.retrieveFlux((Class) expectedResponseType)); + } + else { + return Mono.just(responseSpec.retrieveFlux((ParameterizedTypeReference) expectedResponseType)); + } + default: + throw new UnsupportedOperationException("Unsupported command: " + command); + } + } + + private Object evaluateExpressionForType(Message requestMessage, Expression expression, String propertyName) { + Object type = expression.getValue(this.evaluationContext, requestMessage); + Assert.state(type instanceof Class + || type instanceof String + || type instanceof ParameterizedTypeReference, + () -> "The '" + propertyName + "' [" + expression + + "] must evaluate to 'String' (class FQN), 'Class' " + + "or 'ParameterizedTypeReference', not to: " + type); + + if (type instanceof String) { + try { + return ClassUtils.forName((String) type, getBeanClassLoader()); + } + catch (ClassNotFoundException e) { + throw new IllegalStateException(e); + } + } + else { + return type; + } + } + + /** + * Enumeration of commands supported by the gateways. + */ + public enum Command { + + /** + * Perform {@link RSocket#fireAndForget fireAndForget}. + * @see RSocketRequester.ResponseSpec#send() + */ + fireAndForget, + + /** + * Perform {@link RSocket#requestResponse requestResponse}. + * @see RSocketRequester.ResponseSpec#retrieveMono + */ + requestResponse, + + /** + * Perform {@link RSocket#requestStream requestStream} or + * {@link RSocket#requestChannel requestChannel} depending on whether + * the request input consists of a single or multiple payloads. + * @see RSocketRequester.ResponseSpec#retrieveFlux + */ + requestStreamOrChannel + + } + +} diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/package-info.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/package-info.java new file mode 100644 index 0000000000..06ae5bd659 --- /dev/null +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes representing outbound RSocket components. + */ +package org.springframework.integration.rsocket.outbound; diff --git a/spring-integration-rsocket/src/main/resources/META-INF/spring.handlers b/spring-integration-rsocket/src/main/resources/META-INF/spring.handlers new file mode 100644 index 0000000000..f17bd16a1c --- /dev/null +++ b/spring-integration-rsocket/src/main/resources/META-INF/spring.handlers @@ -0,0 +1 @@ +http\://www.springframework.org/schema/integration/rsocket=org.springframework.integration.rsocket.config.RSocketNamespaceHandler diff --git a/spring-integration-rsocket/src/main/resources/META-INF/spring.schemas b/spring-integration-rsocket/src/main/resources/META-INF/spring.schemas new file mode 100644 index 0000000000..b7a7a6b25d --- /dev/null +++ b/spring-integration-rsocket/src/main/resources/META-INF/spring.schemas @@ -0,0 +1,2 @@ +http\://www.springframework.org/schema/integration/rsocket/spring-integration-rsocket-5.2.xsd=org/springframework/integration/rsocket/config/spring-integration-rsocket-5.2.xsd +http\://www.springframework.org/schema/integration/rsocket/spring-integration-rsocket.xsd=org/springframework/integration/rsocket/config/spring-integration-rsocket-5.2.xsd diff --git a/spring-integration-rsocket/src/main/resources/META-INF/spring.tooling b/spring-integration-rsocket/src/main/resources/META-INF/spring.tooling new file mode 100644 index 0000000000..d1a918ab7d --- /dev/null +++ b/spring-integration-rsocket/src/main/resources/META-INF/spring.tooling @@ -0,0 +1,4 @@ +# Tooling related information for the integration rsocket namespace +http\://www.springframework.org/schema/integration/rsocket@name=integration rsocket Namespace +http\://www.springframework.org/schema/integration/rsocket@prefix=int-rsocket +http\://www.springframework.org/schema/integration/rsocket@icon=org/springframework/integration/rsocket/config/spring-integration-rsocket.gif diff --git a/spring-integration-rsocket/src/main/resources/org/springframework/integration/rsocket/config/spring-integration-rsocket-5.2.xsd b/spring-integration-rsocket/src/main/resources/org/springframework/integration/rsocket/config/spring-integration-rsocket-5.2.xsd new file mode 100644 index 0000000000..1054e74df4 --- /dev/null +++ b/spring-integration-rsocket/src/main/resources/org/springframework/integration/rsocket/config/spring-integration-rsocket-5.2.xsd @@ -0,0 +1,21 @@ + + + + + + + + + + + + diff --git a/spring-integration-rsocket/src/main/resources/org/springframework/integration/rsocket/config/spring-integration-rsocket.gif b/spring-integration-rsocket/src/main/resources/org/springframework/integration/rsocket/config/spring-integration-rsocket.gif new file mode 100644 index 0000000000000000000000000000000000000000..750667e608fcaf4e40253ddf3505d59c2bdfcb44 GIT binary patch literal 578 zcmV-I0=@l5Nk%w1VGsZi0OkMy00030|NnVQPjYW|Wnyk~b9G-}dtGRXTXMBraiCav zyjgj5Kt;tik*L|a|uF=~>vd?g}#YDHzV2F`@y~K0D#e2uhjKRf}xwW*~)QiKor@*_l z?$(5^oZ#x?((BIj{Oq&ReW=84gOF&XxOS$jmWhgdo19Cjt7@>Xn5U+aiHLc8dR?!q zk?7yC;@!l|y^Gkyj?ur2>DHj<;LN_ER@uak;Lf7!-oNg`V(ihB?Z=P$Rm6Q$pJqk2bx_8D zSh;;%YCb2mbW^f)RI6=JvTj(ZY+9#fR+(H;mQ+&z|Nj6000000A^8LW004aeEC2ui z01yBW000M{fPI34goTEOh<#2}R!xb5XJ158Z(WH`L}YndVsTrDQ(j9~YH4V2ZHHKR zV@zOsb8=;eMto~-N_2X5Y=~KAb$2c)FDfV;hg@MAAW=nCEh8B=h7Bb#P&Gd*8xuFx zA}~NRGC3p*27M6_4ebpbCnhvFJqiy6eMv}1LPJ3;Av|CZAR)m3AL`h-yYPWR1q2KT Q?096s!2upR1O);BJ6#qKw*UYD literal 0 HcmV?d00001 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 new file mode 100644 index 0000000000..99d1539fa7 --- /dev/null +++ b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java @@ -0,0 +1,439 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.rsocket.outbound; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.time.Duration; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +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; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.MessageChannels; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.handler.annotation.MessageExceptionHandler; +import org.springframework.messaging.handler.annotation.MessageMapping; +import org.springframework.messaging.rsocket.MessageHandlerAcceptor; +import org.springframework.messaging.rsocket.RSocketStrategies; +import org.springframework.messaging.support.ErrorMessage; +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.RSocketFactory; +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; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.ReplayProcessor; +import reactor.netty.tcp.TcpServer; +import reactor.test.StepVerifier; + +/** + * @author Artem Bilan + * + * @since 5.2 + */ +@SpringJUnitConfig +@DirtiesContext +public class RSocketOutboundGatewayIntegrationTests { + + private static final String ROUTE_HEADER = "rsocket_route"; + + private static final String COMMAND_HEADER = "rsocket_command"; + + private static AnnotationConfigApplicationContext context; + + private static int port; + + private static CloseableChannel server; + + @Autowired + private FluxMessageChannel inputChannel; + + @Autowired + private FluxMessageChannel resultChannel; + + @Autowired + private PollableChannel errorChannel; + + @BeforeAll + static void setup() { + context = new AnnotationConfigApplicationContext(ServerConfig.class); + TcpServer tcpServer = + TcpServer.create().port(0) + .doOnBound(server -> port = server.port()); + server = RSocketFactory.receive() + .frameDecoder(PayloadDecoder.ZERO_COPY) + .acceptor(context.getBean(MessageHandlerAcceptor.class)) + .transport(TcpServerTransport.create(tcpServer)) + .start() + .block(); + } + + @AfterAll + static void tearDown() { + context.close(); + server.dispose(); + } + + @Test + void fireAndForget() { + Disposable disposable = Flux.from(this.resultChannel).subscribe(); + this.inputChannel.send( + MessageBuilder.withPayload("Hello") + .setHeader(ROUTE_HEADER, "receive") + .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.fireAndForget) + .build()); + + StepVerifier.create(context.getBean(ServerController.class).fireForgetPayloads) + .expectNext("Hello") + .thenCancel() + .verify(); + + disposable.dispose(); + } + + @Test + void echo() { + this.inputChannel.send( + MessageBuilder.withPayload("Hello") + .setHeader(ROUTE_HEADER, "echo") + .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse) + .build()); + + StepVerifier.create( + Flux.from(this.resultChannel) + .map(Message::getPayload) + .cast(String.class)) + .expectNext("Hello") + .thenCancel() + .verify(); + } + + @Test + void echoAsync() { + this.inputChannel.send( + MessageBuilder.withPayload("Hello") + .setHeader(ROUTE_HEADER, "echo-async") + .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse) + .build()); + + StepVerifier.create( + Flux.from(this.resultChannel) + .map(Message::getPayload) + .cast(String.class)) + .expectNext("Hello async") + .thenCancel() + .verify(); + } + + @Test + void echoStream() { + this.inputChannel.send( + MessageBuilder.withPayload("Hello") + .setHeader(ROUTE_HEADER, "echo-stream") + .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestStreamOrChannel) + .build()); + + Message resultMessage = + Flux.from(this.resultChannel) + .blockFirst(); + + assertThat(resultMessage) + .isNotNull() + .extracting(Message::getPayload) + .isInstanceOf(Flux.class); + + @SuppressWarnings("unchecked") + Flux resultStream = (Flux) resultMessage.getPayload(); + StepVerifier.create(resultStream) + .expectNext("Hello 0").expectNextCount(6).expectNext("Hello 7") + .thenCancel() + .verify(); + + } + + @Test + void echoChannel() { + this.inputChannel.send( + MessageBuilder.withPayload(Flux.range(1, 10).map(i -> "Hello " + i)) + .setHeader(ROUTE_HEADER, "echo-channel") + .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestStreamOrChannel) + .build()); + + Message resultMessage = + Flux.from(this.resultChannel) + .blockFirst(); + + assertThat(resultMessage) + .isNotNull() + .extracting(Message::getPayload) + .isInstanceOf(Flux.class); + + @SuppressWarnings("unchecked") + Flux resultStream = (Flux) resultMessage.getPayload(); + StepVerifier.create(resultStream) + .expectNext("Hello 1 async").expectNextCount(8).expectNext("Hello 10 async") + .thenCancel() + .verify(); + } + + @Test + void voidReturnValue() { + this.inputChannel.send( + MessageBuilder.withPayload("Hello") + .setHeader(ROUTE_HEADER, "void-return-value") + .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestStreamOrChannel) + .build()); + + Message resultMessage = + Flux.from(this.resultChannel) + .blockFirst(); + + assertThat(resultMessage) + .isNotNull() + .extracting(Message::getPayload) + .isInstanceOf(Flux.class); + + Flux resultStream = (Flux) resultMessage.getPayload(); + StepVerifier.create(resultStream) + .expectComplete() + .verify(); + } + + @Test + void voidReturnValueFromExceptionHandler() { + this.inputChannel.send( + MessageBuilder.withPayload("bad") + .setHeader(ROUTE_HEADER, "void-return-value") + .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestStreamOrChannel) + .build()); + + Message resultMessage = + Flux.from(this.resultChannel) + .blockFirst(); + + assertThat(resultMessage) + .isNotNull() + .extracting(Message::getPayload) + .isInstanceOf(Flux.class); + + Flux resultStream = (Flux) resultMessage.getPayload(); + StepVerifier.create(resultStream) + .expectComplete() + .verify(); + } + + @Test + void handleWithThrownException() { + this.inputChannel.send( + MessageBuilder.withPayload("a") + .setHeader(ROUTE_HEADER, "thrown-exception") + .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse) + .build()); + + StepVerifier.create( + Flux.from(this.resultChannel) + .map(Message::getPayload) + .cast(String.class)) + .expectNext("Invalid input error handled") + .thenCancel() + .verify(); + } + + @Test + void handleWithErrorSignal() { + this.inputChannel.send( + MessageBuilder.withPayload("a") + .setHeader(ROUTE_HEADER, "error-signal") + .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse) + .build()); + + StepVerifier.create( + Flux.from(this.resultChannel) + .map(Message::getPayload) + .cast(String.class)) + .expectNext("Invalid input error handled") + .thenCancel() + .verify(); + } + + @Test + void noMatchingRoute() { + Disposable disposable = Flux.from(this.resultChannel).subscribe(); + this.inputChannel.send( + MessageBuilder.withPayload("anything") + .setHeader(ROUTE_HEADER, "invalid") + .setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse) + .build()); + + Message errorMessage = errorChannel.receive(10_000); + + assertThat(errorMessage).isNotNull() + .isInstanceOf(ErrorMessage.class) + .extracting(Message::getPayload) + .isInstanceOf(MessageHandlingException.class) + .satisfies((ex) -> assertThat((Exception) ex) + .hasMessageContaining("io.rsocket.exceptions.ApplicationErrorException: " + + "No handler for destination 'invalid'")); + + disposable.dispose(); + } + + @Configuration + @EnableIntegration + public static class ClientConfig { + + @Bean + public MessageHandler rsocketOutboundGateway() { + RSocketOutboundGateway rsocketOutboundGateway = + new RSocketOutboundGateway(TcpClientTransport.create(port), + new FunctionExpression>((m) -> m.getHeaders().get(ROUTE_HEADER))); + rsocketOutboundGateway.setCommandExpression( + new FunctionExpression>((m) -> m.getHeaders().get(COMMAND_HEADER))); + rsocketOutboundGateway.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY)); + rsocketOutboundGateway.setStrategiesConfigurer((strategies) -> + strategies.decoder(StringDecoder.allMimeTypes()) + .encoder(CharSequenceEncoder.allMimeTypes()) + .dataBufferFactory(new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT))); + return rsocketOutboundGateway; + } + + @Bean + public IntegrationFlow rsocketOutboundFlow() { + return IntegrationFlows.from(MessageChannels.flux("inputChannel")) + .handle(rsocketOutboundGateway()) + .channel(c -> c.flux("resultChannel")) + .get(); + } + + @Bean + public PollableChannel errorChannel() { + return new QueueChannel(); + } + + } + + @Configuration + static class ServerConfig { + + @Bean + public ServerController controller() { + return new ServerController(); + } + + @Bean + public MessageHandlerAcceptor messageHandlerAcceptor() { + MessageHandlerAcceptor acceptor = new MessageHandlerAcceptor(); + acceptor.setRSocketStrategies(rsocketStrategies()); + return acceptor; + } + + @Bean + public RSocketStrategies rsocketStrategies() { + return RSocketStrategies.builder() + .decoder(StringDecoder.allMimeTypes()) + .encoder(CharSequenceEncoder.allMimeTypes()) + .dataBufferFactory(new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT)) + .build(); + } + + } + + @Controller + static class ServerController { + + final ReplayProcessor fireForgetPayloads = ReplayProcessor.create(); + + @MessageMapping("receive") + void receive(String payload) { + this.fireForgetPayloads.onNext(payload); + } + + @MessageMapping("echo") + String echo(String payload) { + return payload; + } + + @MessageMapping("echo-async") + Mono echoAsync(String payload) { + return Mono.delay(Duration.ofMillis(10)).map(aLong -> payload + " async"); + } + + @MessageMapping("echo-stream") + Flux echoStream(String payload) { + return Flux.interval(Duration.ofMillis(10)).map(aLong -> payload + " " + aLong); + } + + @MessageMapping("echo-channel") + Flux echoChannel(Flux payloads) { + return payloads.delayElements(Duration.ofMillis(10)).map(payload -> payload + " async"); + } + + @MessageMapping("thrown-exception") + Mono handleAndThrow(String payload) { + throw new IllegalArgumentException("Invalid input error"); + } + + @MessageMapping("error-signal") + Mono handleAndReturnError(String payload) { + return Mono.error(new IllegalArgumentException("Invalid input error")); + } + + @MessageMapping("void-return-value") + Mono voidReturnValue(String payload) { + return !payload.equals("bad") ? + Mono.delay(Duration.ofMillis(10)).then(Mono.empty()) : + Mono.error(new IllegalStateException("bad")); + } + + @MessageExceptionHandler + Mono handleException(IllegalArgumentException ex) { + return Mono.delay(Duration.ofMillis(10)).map(aLong -> ex.getMessage() + " handled"); + } + + @MessageExceptionHandler + Mono handleExceptionWithVoidReturnValue(IllegalStateException ex) { + return Mono.delay(Duration.ofMillis(10)).then(Mono.empty()); + } + + } + +} diff --git a/spring-integration-rsocket/src/test/resources/log4j2-test.xml b/spring-integration-rsocket/src/test/resources/log4j2-test.xml new file mode 100644 index 0000000000..d6397df41c --- /dev/null +++ b/spring-integration-rsocket/src/test/resources/log4j2-test.xml @@ -0,0 +1,16 @@ + + + + + + + + + + + + + + + +