Add RSocket Java DSL

* Add `@NonNullApi` for RSocket packages
* Some code style and JavaDocs polishing

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) {

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

@@ -161,7 +161,7 @@ public class ClientRSocketConnector extends AbstractRSocketConnector {
 
 	public Mono<RSocketRequester> getRSocketRequester() {
 		return this.rsocketMono
-				.map(rsocket -> RSocketRequester.wrap(rsocket, getDataMimeType(), getRSocketStrategies()))
+				.map((rsocket) -> RSocketRequester.wrap(rsocket, getDataMimeType(), getRSocketStrategies()))
 				.cache();
 	}
 

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<DataBuffer> buffers = payloads.map(this::retainDataAndReleasePayload).doOnSubscribe(s -> read.set(true));
+		Flux<DataBuffer> buffers =
+				payloads.map(this::retainDataAndReleasePayload)
+						.doOnSubscribe((subscription) -> read.set(true));
 		Message<Flux<DataBuffer>> 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) {

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;

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, RSocketInboundGateway> {
+
+	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;
+	}
+
+}

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, RSocketOutboundGateway> {
+
+	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 <P> the expected request message payload type.
+	 * @return the spec
+	 * @see RSocketOutboundGateway#setCommandExpression(Expression)
+	 */
+	public <P> RSocketOutboundGatewaySpec command(Function<Message<P>, ?> 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 <P> the expected request message payload type.
+	 * @return the spec
+	 * @see RSocketOutboundGateway#setPublisherElementTypeExpression(Expression)
+	 */
+	public <P> RSocketOutboundGatewaySpec publisherElementType(Function<Message<P>, ?> 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 <P> the expected request message payload type.
+	 * @return the spec
+	 * @see RSocketOutboundGateway#setExpectedResponseTypeExpression(Expression)
+	 */
+	public <P> RSocketOutboundGatewaySpec expectedResponseType(Function<Message<P>, ?> 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;
+	}
+
+}

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 <P> the expected payload type.
+	 * @return the RSocketOutboundGatewaySpec instance
+	 */
+	public static <P> RSocketOutboundGatewaySpec outboundGateway(Function<Message<P>, ?> 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();
+	}
+
+}

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;

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 <code>byte[].class</code>.
+	 * 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 <code>byte[].class</code>.
+	 * 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)

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;

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) {

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;

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;

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<String, String> 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()))
+					.<Flux<String>, Mono<String>>transform((flux) -> flux.next().map(String::toUpperCase))
+					.get();
+		}
+
+	}
+
+}