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 09d32af538..068d48bfae 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 @@ -59,6 +59,8 @@ public abstract class AbstractRSocketConnector .dataBufferFactory(new DefaultDataBufferFactory()) .build(); + private boolean autoStartup = true; + private volatile boolean running; protected AbstractRSocketConnector(IntegrationRSocketAcceptor rsocketAcceptor) { @@ -128,6 +130,15 @@ public abstract class AbstractRSocketConnector this.rsocketAcceptor.detectEndpoints(); } + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + @Override + public boolean isAutoStartup() { + return this.autoStartup; + } + @Override public void start() { if (!this.running) { 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 0a9cb8e298..9b75b80aba 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 @@ -161,7 +161,7 @@ public class ClientRSocketConnector extends AbstractRSocketConnector { public Mono getRSocketRequester() { return this.rsocketMono - .map(rsocket -> RSocketRequester.wrap(rsocket, getDataMimeType(), getRSocketStrategies())) + .map((rsocket) -> RSocketRequester.wrap(rsocket, getDataMimeType(), getRSocketStrategies())) .cache(); } diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java index f900f3af89..ede9e1a2c1 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java @@ -129,7 +129,7 @@ class IntegrationRSocket extends AbstractRSocket { int refCount = refCount(dataBuffer); Message message = MessageBuilder.createMessage(dataBuffer, headers); return Mono.defer(() -> this.handler.apply(message)) - .doFinally(s -> { + .doFinally((signal) -> { if (refCount(dataBuffer) == refCount) { DataBufferUtils.release(dataBuffer); } @@ -147,19 +147,22 @@ class IntegrationRSocket extends AbstractRSocket { MessageHeaders headers = createHeaders(destination, replyMono); AtomicBoolean read = new AtomicBoolean(); - Flux buffers = payloads.map(this::retainDataAndReleasePayload).doOnSubscribe(s -> read.set(true)); + Flux buffers = + payloads.map(this::retainDataAndReleasePayload) + .doOnSubscribe((subscription) -> read.set(true)); Message> message = MessageBuilder.createMessage(buffers, headers); return Mono.defer(() -> this.handler.apply(message)) - .doFinally(s -> { + .doFinally((signal) -> { // Subscription should have happened by now due to ChannelSendOperator if (!read.get()) { buffers.subscribe(DataBufferUtils::release); } }) - .thenMany(Flux.defer(() -> replyMono.isTerminated() ? - replyMono.flatMapMany(Function.identity()) : - Mono.error(new IllegalStateException("Something went wrong: reply Mono not set")))); + .thenMany(Flux.defer(() -> + replyMono.isTerminated() + ? replyMono.flatMapMany(Function.identity()) + : Mono.error(new IllegalStateException("Something went wrong: reply Mono not set")))); } private DataBuffer retainDataAndReleasePayload(Payload payload) { diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/package-info.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/package-info.java index 8dcdba113b..a598aa16c2 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/package-info.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/package-info.java @@ -1,4 +1,5 @@ /** * Provides classes for RSocket XML namespace parsing and configuration support. */ +@org.springframework.lang.NonNullApi package org.springframework.integration.rsocket.config; diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSocketInboundGatewaySpec.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSocketInboundGatewaySpec.java new file mode 100644 index 0000000000..ce5eaa5bf6 --- /dev/null +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSocketInboundGatewaySpec.java @@ -0,0 +1,72 @@ +/* + * 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.dsl; + +import org.springframework.core.ResolvableType; +import org.springframework.integration.dsl.MessagingGatewaySpec; +import org.springframework.integration.rsocket.AbstractRSocketConnector; +import org.springframework.integration.rsocket.inbound.RSocketInboundGateway; +import org.springframework.messaging.rsocket.RSocketStrategies; + +/** + * The {@link MessagingGatewaySpec} implementation for the {@link RSocketInboundGateway}. + * + * @author Artem Bilan + * + * @since 5.2 + */ +public class RSocketInboundGatewaySpec extends MessagingGatewaySpec { + + RSocketInboundGatewaySpec(String... path) { + super(new RSocketInboundGateway(path)); + } + + /** + * Configure {@link RSocketStrategies} instead of a default one. + * @param rsocketStrategies the {@link RSocketStrategies} to use. + * @return the spec + * @see RSocketInboundGateway#setRSocketStrategies(RSocketStrategies) + */ + public RSocketInboundGatewaySpec rsocketStrategies(RSocketStrategies rsocketStrategies) { + this.target.setRSocketStrategies(rsocketStrategies); + return this; + } + + /** + * Provide an {@link AbstractRSocketConnector} reference for an explicit endpoint mapping. + * @param rsocketConnector the {@link AbstractRSocketConnector} to use. + * @return the spec + * @see RSocketInboundGateway#setRSocketConnector(AbstractRSocketConnector) + */ + public RSocketInboundGatewaySpec rsocketConnector(AbstractRSocketConnector rsocketConnector) { + this.target.setRSocketConnector(rsocketConnector); + return this; + } + + /** + * Specify the type of payload to be generated when the inbound RSocket request + * content is read by the converters/encoders. + * @param requestElementType The payload type. + * @return the spec + * @see RSocketInboundGateway#setRequestElementType(ResolvableType) + */ + public RSocketInboundGatewaySpec requestElementType(ResolvableType requestElementType) { + this.target.setRequestElementType(requestElementType); + return this; + } + +} diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSocketOutboundGatewaySpec.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSocketOutboundGatewaySpec.java new file mode 100644 index 0000000000..b5218440f6 --- /dev/null +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSocketOutboundGatewaySpec.java @@ -0,0 +1,195 @@ +/* + * 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.dsl; + +import java.util.function.Function; + +import org.reactivestreams.Publisher; + +import org.springframework.expression.Expression; +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.expression.ValueExpression; +import org.springframework.integration.rsocket.ClientRSocketConnector; +import org.springframework.integration.rsocket.outbound.RSocketOutboundGateway; +import org.springframework.messaging.Message; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketRequesterMethodArgumentResolver; + +/** + * The {@link MessageHandlerSpec} implementation for the {@link RSocketOutboundGateway}. + * + * @author Artem Bilan + * + * @since 5.2 + */ +public class RSocketOutboundGatewaySpec extends MessageHandlerSpec { + + RSocketOutboundGatewaySpec(Expression routeExpression) { + this.target = new RSocketOutboundGateway(routeExpression); + } + + /** + * Configure a {@link ClientRSocketConnector} for client side requests based on the connection + * provided by the {@link ClientRSocketConnector#getRSocketRequester()}. + * In case of server side, an {@link RSocketRequester} must be provided in the + * {@link RSocketRequesterMethodArgumentResolver#RSOCKET_REQUESTER_HEADER} header of request message. + * @param clientRSocketConnector the {@link ClientRSocketConnector} to use. + * @return the spec + * @see RSocketOutboundGateway#setClientRSocketConnector(ClientRSocketConnector) + */ + public RSocketOutboundGatewaySpec clientRSocketConnector(ClientRSocketConnector clientRSocketConnector) { + this.target.setClientRSocketConnector(clientRSocketConnector); + return this; + } + + /** + * Configure a {@link RSocketOutboundGateway.Command} for RSocket request type. + * @param command the {@link RSocketOutboundGateway.Command} to use. + * @return the spec + * @see RSocketOutboundGateway#setCommand(RSocketOutboundGateway.Command) + */ + public RSocketOutboundGatewaySpec command(RSocketOutboundGateway.Command command) { + return command(new ValueExpression<>(command)); + } + + /** + * Configure a {@code Function} to evaluate a {@link RSocketOutboundGateway.Command} + * for RSocket request type at runtime against a request message. + * @param commandFunction the {@code Function} to use. + * @param

the expected request message payload type. + * @return the spec + * @see RSocketOutboundGateway#setCommandExpression(Expression) + */ + public

RSocketOutboundGatewaySpec command(Function, ?> commandFunction) { + return command(new FunctionExpression<>(commandFunction)); + } + + /** + * Configure a SpEL expression to evaluate a {@link RSocketOutboundGateway.Command} + * for RSocket request type at runtime against a request message. + * @param commandExpression the SpEL expression to use. + * @return the spec + * @see RSocketOutboundGateway#setCommandExpression(Expression) + */ + public RSocketOutboundGatewaySpec command(String commandExpression) { + return command(PARSER.parseExpression(commandExpression)); + } + + /** + * Configure a SpEL expression to evaluate a {@link RSocketOutboundGateway.Command} + * for RSocket request type at runtime against a request message. + * @param commandExpression the SpEL expression to use. + * @return the spec + * @see RSocketOutboundGateway#setCommandExpression(Expression) + */ + public RSocketOutboundGatewaySpec command(Expression commandExpression) { + this.target.setCommandExpression(commandExpression); + return this; + } + + /** + * Configure a type for a request {@link Publisher} elements. + * @param publisherElementType the type of the request {@link Publisher} elements. + * @return the spec + * @see RSocketOutboundGateway#setPublisherElementType(Class) + */ + public RSocketOutboundGatewaySpec publisherElementType(Class publisherElementType) { + return publisherElementType(new ValueExpression<>(publisherElementType)); + } + + /** + * Configure a {@code Function} to evaluate a request {@link Publisher} elements type at runtime against + * a request message. + * @param publisherElementTypeFunction the {@code Function} to evaluate a type for the request + * {@link Publisher} elements. + * @param

the expected request message payload type. + * @return the spec + * @see RSocketOutboundGateway#setPublisherElementTypeExpression(Expression) + */ + public

RSocketOutboundGatewaySpec publisherElementType(Function, ?> publisherElementTypeFunction) { + return publisherElementType(new FunctionExpression<>(publisherElementTypeFunction)); + } + + /** + * 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. + * @return the spec + * @see RSocketOutboundGateway#setPublisherElementTypeExpression(Expression) + */ + public RSocketOutboundGatewaySpec publisherElementType(String publisherElementTypeExpression) { + return publisherElementType(PARSER.parseExpression(publisherElementTypeExpression)); + } + + /** + * 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. + * @return the spec + * @see RSocketOutboundGateway#setPublisherElementTypeExpression(Expression) + */ + public RSocketOutboundGatewaySpec publisherElementType(Expression publisherElementTypeExpression) { + this.target.setPublisherElementTypeExpression(publisherElementTypeExpression); + return this; + } + + /** + * Specify the expected response type for the RSocket response. + * @param expectedResponseType The expected type. + * @return the spec + * @see RSocketOutboundGateway#setExpectedResponseType(Class) + */ + public RSocketOutboundGatewaySpec expectedResponseType(Class expectedResponseType) { + return expectedResponseType(new ValueExpression<>(expectedResponseType)); + } + + /** + * Specify the {@code Function} to determine the type for the RSocket response. + * @param expectedResponseTypeFunction The expected response type {@code Function}. + * @param

the expected request message payload type. + * @return the spec + * @see RSocketOutboundGateway#setExpectedResponseTypeExpression(Expression) + */ + public

RSocketOutboundGatewaySpec expectedResponseType(Function, ?> expectedResponseTypeFunction) { + return expectedResponseType(new FunctionExpression<>(expectedResponseTypeFunction)); + } + + /** + * Specify the {@link Expression} to determine the type for the RSocket response. + * @param expectedResponseTypeExpression The expected response type expression. + * @return the spec + * @see RSocketOutboundGateway#setExpectedResponseTypeExpression(Expression) + */ + public RSocketOutboundGatewaySpec expectedResponseType(String expectedResponseTypeExpression) { + return expectedResponseType(PARSER.parseExpression(expectedResponseTypeExpression)); + } + + /** + * Specify the {@link Expression} to determine the type for the RSocket response. + * @param expectedResponseTypeExpression The expected response type expression. + * @return the spec + * @see RSocketOutboundGateway#setExpectedResponseTypeExpression(Expression) + */ + public RSocketOutboundGatewaySpec expectedResponseType(Expression expectedResponseTypeExpression) { + this.target.setExpectedResponseTypeExpression(expectedResponseTypeExpression); + return this; + } + +} diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSockets.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSockets.java new file mode 100644 index 0000000000..6a46c78724 --- /dev/null +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSockets.java @@ -0,0 +1,80 @@ +/* + * 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.dsl; + +import java.util.function.Function; + +import org.springframework.expression.Expression; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.messaging.Message; + +/** + * The RSocket components Factory. + * + * @author Artem Bilan + * + * @since 5.2 + */ +public final class RSockets { + + /** + * Create an {@link RSocketOutboundGatewaySpec} builder for request-reply gateway + * based on provided {@code route}. + * @param route the {@code route} to send requests. + * @return the RSocketOutboundGatewaySpec instance + */ + public static RSocketOutboundGatewaySpec outboundGateway(String route) { + return outboundGateway(new LiteralExpression(route)); + } + + /** + * Create an {@link RSocketOutboundGatewaySpec} builder for request-reply gateway + * based on provided {@code Function} to evaluate target {@code route} against request message. + * @param routeFunction the {@code Function} to evaluate {@code route} at runtime. + * @param

the expected payload type. + * @return the RSocketOutboundGatewaySpec instance + */ + public static

RSocketOutboundGatewaySpec outboundGateway(Function, ?> routeFunction) { + return outboundGateway(new FunctionExpression<>(routeFunction)); + } + + /** + * Create an {@link RSocketOutboundGatewaySpec} builder for request-reply gateway + * based on provided SpEL {@link Expression} to evaluate target {@code route} against request message. + * @param routeExpression the SpEL {@link Expression} to evaluate {@code route} at runtime. + * @return the RSocketOutboundGatewaySpec instance + */ + public static RSocketOutboundGatewaySpec outboundGateway(Expression routeExpression) { + return new RSocketOutboundGatewaySpec(routeExpression); + } + + /** + * Create an {@link RSocketInboundGatewaySpec} builder for request-reply reactive gateway + * based on the provided {@code path} array for mapping. + * @param path the path mapping URIs (e.g. "/myPath.do"). + * @return the RSocketInboundGatewaySpec instance + */ + public static RSocketInboundGatewaySpec inboundGateway(String... path) { + return new RSocketInboundGatewaySpec(path); + } + + private RSockets() { + super(); + } + +} diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/package-info.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/package-info.java new file mode 100644 index 0000000000..28166a8732 --- /dev/null +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/package-info.java @@ -0,0 +1,5 @@ +/** + * Provides RSocket Components support for Spring Integration Java DSL. + */ +@org.springframework.lang.NonNullApi +package org.springframework.integration.rsocket.dsl; diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java index b5e2d2b8dc..4cc669c55e 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java @@ -104,7 +104,7 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In /** * Configure {@link RSocketStrategies} instead of a default one. - * Note: if {@link AbstractRSocketConnector} ias provided, then its + * Note: if {@link AbstractRSocketConnector} is provided, then its * {@link RSocketStrategies} have a precedence. * @param rsocketStrategies the {@link RSocketStrategies} to use. * @see RSocketStrategies#builder @@ -135,7 +135,7 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In * Specify the type of payload to be generated when the inbound RSocket request * content is read by the encoders. * By default this value is null which means at runtime any "text" Content-Type will - * result in String while all others default to byte[].class. + * result in String while all others default to {@code byte[].class}. * @param requestElementClass The payload type. */ public void setRequestElementClass(Class requestElementClass) { @@ -146,7 +146,7 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In * Specify the type of payload to be generated when the inbound RSocket request * content is read by the converters/encoders. * By default this value is null which means at runtime any "text" Content-Type will - * result in String while all others default to byte[].class. + * result in String while all others default to {@code byte[].class}. * @param requestElementType The payload type. */ public void setRequestElementType(ResolvableType requestElementType) { @@ -234,6 +234,10 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In requestMessageHeaders.get(HandlerMethodReturnValueHandler.DATA_BUFFER_FACTORY_HEADER, DataBufferFactory.class); + if (bufferFactory == null) { + bufferFactory = this.rsocketStrategies.dataBufferFactory(); + } + MimeType mimeType = requestMessageHeaders.get(MessageHeaders.CONTENT_TYPE, MimeType.class); return encodeContent(reply, ResolvableType.forInstance(reply), bufferFactory, mimeType) diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/package-info.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/package-info.java index b87d225241..e2d1a45fbb 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/package-info.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/package-info.java @@ -1,4 +1,5 @@ /** * Provides classes representing inbound RSocket components. */ +@org.springframework.lang.NonNullApi package org.springframework.integration.rsocket.inbound; 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 index 1ac5bbc38b..a3865596a3 100644 --- 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 @@ -137,7 +137,6 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler * @see RSocketRequester.RequestSpec#data(Publisher, Class) */ public void setPublisherElementType(Class publisherElementType) { - Assert.notNull(publisherElementType, "'publisherElementType' must not be null"); setPublisherElementTypeExpression(new ValueExpression<>(publisherElementType)); } @@ -161,7 +160,6 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler * @see RSocketRequester.ResponseSpec#retrieveFlux */ public void setExpectedResponseType(Class expectedResponseType) { - Assert.notNull(expectedResponseType, "'expectedResponseType' must not be null"); setExpectedResponseTypeExpression(new ValueExpression<>(expectedResponseType)); } @@ -222,7 +220,7 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler Object payload = requestMessage.getPayload(); if (payload instanceof Publisher && this.publisherElementTypeExpression != null) { Object publisherElementType = evaluateExpressionForType(requestMessage, this.publisherElementTypeExpression, - "publisherElementTypeExpression"); + "publisherElementType"); return responseSpecForPublisher(requestSpec, (Publisher) payload, publisherElementType); } else { @@ -245,12 +243,12 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler 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"); + () -> "The 'command' [" + this.commandExpression + "] must not evaluate to null"); Object expectedResponseType = null; if (!Command.fireAndForget.equals(command)) { expectedResponseType = evaluateExpressionForType(requestMessage, this.expectedResponseTypeExpression, - "expectedResponseTypeExpression"); + "expectedResponseType"); } switch (command) { 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 index 06ae5bd659..10b1e5b739 100644 --- 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 @@ -1,4 +1,5 @@ /** * Provides classes representing outbound RSocket components. */ +@org.springframework.lang.NonNullApi package org.springframework.integration.rsocket.outbound; diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/package-info.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/package-info.java index 1ab963c7af..6294cd076e 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/package-info.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/package-info.java @@ -1,4 +1,5 @@ /** * Provides common classes for RSocket components. */ +@org.springframework.lang.NonNullApi package org.springframework.integration.rsocket; diff --git a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java new file mode 100644 index 0000000000..4efafc1a34 --- /dev/null +++ b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java @@ -0,0 +1,140 @@ +/* + * 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.dsl; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.function.Function; + +import org.junit.jupiter.api.Test; + +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.context.annotation.DependsOn; +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.netty.util.NetUtil; +import io.rsocket.frame.decoder.PayloadDecoder; +import io.rsocket.transport.netty.client.TcpClientTransport; +import io.rsocket.transport.netty.server.TcpServerTransport; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.netty.tcp.InetSocketAddressUtil; +import reactor.netty.tcp.TcpClient; +import reactor.netty.tcp.TcpServer; + +/** + * @author Artem Bilan + * + * @since 5.2 + */ +@SpringJUnitConfig +@DirtiesContext +public class RSocketDslTests { + + @Autowired + @Qualifier("rsocketUpperCaseRequestFlow.gateway") + private Function rsocketUpperCaseFlowFunction; + + @Test + void testRsocketUpperCaseFlows() { + assertThat(this.rsocketUpperCaseFlowFunction.apply("hello world")).isEqualTo("HELLO WORLD"); + } + + @Configuration + @EnableIntegration + public static class TestConfiguration { + + private volatile int port; + + @Bean + public RSocketStrategies rsocketStrategies() { + return RSocketStrategies.builder() + .decoder(StringDecoder.allMimeTypes()) + .encoder(CharSequenceEncoder.allMimeTypes()) + .dataBufferFactory(new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT)) + .build(); + } + + @Bean + public ServerRSocketConnector serverRSocketConnector() { + TcpServer tcpServer = + TcpServer.create() + .port(0) + .doOnBound((server) -> this.port = server.port()); + ServerRSocketConnector serverRSocketConnector = + new ServerRSocketConnector(TcpServerTransport.create(tcpServer)); + serverRSocketConnector.setRSocketStrategies(rsocketStrategies()); + serverRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY)); + return serverRSocketConnector; + } + + + @Bean + @DependsOn("serverRSocketConnector") + public ClientRSocketConnector clientRSocketConnector() { + ClientRSocketConnector clientRSocketConnector = + new ClientRSocketConnector( + TcpClientTransport.create( + TcpClient.create() + .addressSupplier(() -> + InetSocketAddressUtil.createUnresolved( + NetUtil.LOCALHOST.getHostAddress(), this.port)) + )); + clientRSocketConnector.setFactoryConfigurer((factory) -> factory.frameDecoder(PayloadDecoder.ZERO_COPY)); + clientRSocketConnector.setRSocketStrategies(rsocketStrategies()); + clientRSocketConnector.setAutoStartup(false); + return clientRSocketConnector; + } + + @Bean + public IntegrationFlow rsocketUpperCaseRequestFlow() { + return IntegrationFlows + .from(Function.class) + .handle(RSockets.outboundGateway("/uppercase") + .command((message) -> RSocketOutboundGateway.Command.requestResponse) + .expectedResponseType("T(java.lang.String)") + .clientRSocketConnector(clientRSocketConnector())) + .get(); + } + + @Bean + public IntegrationFlow rsocketUpperCaseFlow() { + return IntegrationFlows + .from(RSockets.inboundGateway("/uppercase") + .rsocketStrategies(rsocketStrategies())) + ., Mono>transform((flux) -> flux.next().map(String::toUpperCase)) + .get(); + } + + } + +}