Introduce RSocketInteractionModel

* Deprecate `RSocketOutboundGateway.Command` in favor of newly introduced
top level `RSocketInteractionModel`
* Deprecate setter, DSL and XML configurations for the deprecated
`RSocketOutboundGateway.Command` in favor of newly introduced configurators
for the mentioned `RSocketInteractionModel`
* Add `IntegrationRSocketEndpoint.getInteractionModels()` contract for
inbound endpoints.
This way we can restrict mapped endpoints to the particular interaction
model(s)
* Add DSL and XML configuration for inbound interaction model option
* Document changes

Author: artembilan

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

@@ -39,4 +39,15 @@ public interface IntegrationRSocketEndpoint extends ReactiveMessageHandler {
 	 */
 	String[] getPath();
 
+	/**
+	 * Obtain {@link RSocketInteractionModel}s
+	 * this {@link ReactiveMessageHandler} is going to be mapped onto.
+	 * Defaults to all the {@link RSocketInteractionModel}s.
+	 * @return the interaction models for mapping.
+	 * @since 5.2.2
+	 */
+	default RSocketInteractionModel[] getInteractionModels() {
+		return RSocketInteractionModel.values();
+	}
+
 }

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

@@ -17,6 +17,7 @@
 package org.springframework.integration.rsocket;
 
 import java.lang.reflect.Method;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
@@ -79,13 +80,19 @@ public boolean detectEndpoints() {
 	}
 
 	public void addEndpoint(IntegrationRSocketEndpoint endpoint) {
+		RSocketFrameTypeMessageCondition frameTypeMessageCondition = RSocketFrameTypeMessageCondition.EMPTY_CONDITION;
+
+		RSocketInteractionModel[] interactionModels = endpoint.getInteractionModels();
+		if (interactionModels.length > 0) {
+			frameTypeMessageCondition =
+					new RSocketFrameTypeMessageCondition(
+							Arrays.stream(interactionModels)
+									.map(RSocketInteractionModel::getFrameType)
+									.toArray(FrameType[]::new));
+		}
 		registerHandlerMethod(endpoint, HANDLE_MESSAGE_METHOD,
 				new CompositeMessageCondition(
-						new RSocketFrameTypeMessageCondition(
-								FrameType.REQUEST_FNF,
-								FrameType.REQUEST_RESPONSE,
-								FrameType.REQUEST_STREAM,
-								FrameType.REQUEST_CHANNEL),
+						frameTypeMessageCondition,
 						new DestinationPatternsMessageCondition(endpoint.getPath(), getRouteMatcher()))); // NOSONAR
 	}
 

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

@@ -0,0 +1,63 @@
+/*
+ * 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;
+
+import io.rsocket.frame.FrameType;
+
+/**
+ * The RSocket protocol interaction models.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.2.2
+ *
+ * @see <a href="https://rsocket.io/">RSocket protocol official site</a>
+ * @see FrameType
+ */
+public enum RSocketInteractionModel {
+
+	/**
+	 * The model for {@link io.rsocket.RSocket#fireAndForget} operation.
+	 */
+	fireAndForget(FrameType.REQUEST_FNF),
+
+	/**
+	 * The model for {@link io.rsocket.RSocket#requestResponse} operation.
+	 */
+	requestResponse(FrameType.REQUEST_RESPONSE),
+
+	/**
+	 * The model for {@link io.rsocket.RSocket#requestStream} operation.
+	 */
+	requestStream(FrameType.REQUEST_STREAM),
+
+	/**
+	 * The model for {@link io.rsocket.RSocket#requestChannel} operation.
+	 */
+	requestChannel(FrameType.REQUEST_CHANNEL);
+
+	private final FrameType frameType;
+
+	RSocketInteractionModel(FrameType frameType) {
+		this.frameType = frameType;
+	}
+
+	public FrameType getFrameType() {
+		return this.frameType;
+	}
+
+}

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketInboundGatewayParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * 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.
@@ -29,13 +29,15 @@
 /**
  * Parser for the <inbound-gateway/> element of the 'rsocket' namespace.
  *
- * @author Mark Fisher
- * @author Gary Russell
+ * @author Artem Bilan
+ *
+ * @since 5.2
  */
 public class RSocketInboundGatewayParser extends AbstractInboundGatewayParser {
 
 	private static final List<String> NON_ELIGIBLE_ATTRIBUTES =
 			Arrays.asList("path",
+					"interaction-models",
 					"rsocket-strategies",
 					"rsocket-connector",
 					"request-element-type");
@@ -59,6 +61,7 @@ protected void doPostProcess(BeanDefinitionBuilder builder, Element element) {
 				"rSocketStrategies");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "rsocket-connector",
 				"RSocketConnector");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "interaction-models");
 	}
 
 }

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/config/RSocketOutboundGatewayParser.java

@@ -49,6 +49,7 @@ protected BeanDefinitionBuilder parseHandler(Element element, ParserContext pars
 		builder.addConstructorArgValue(routeExpression);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "client-rsocket-connector",
 				"clientRSocketConnector");
+		populateValueOrExpressionIfAny(builder, element, parserContext, "interaction-model");
 		populateValueOrExpressionIfAny(builder, element, parserContext, "command");
 		populateValueOrExpressionIfAny(builder, element, parserContext, "publisher-element-type");
 		populateValueOrExpressionIfAny(builder, element, parserContext, "expected-response-type");

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSocketInboundGatewaySpec.java

@@ -19,6 +19,7 @@
 import org.springframework.core.ResolvableType;
 import org.springframework.integration.dsl.MessagingGatewaySpec;
 import org.springframework.integration.rsocket.AbstractRSocketConnector;
+import org.springframework.integration.rsocket.RSocketInteractionModel;
 import org.springframework.integration.rsocket.inbound.RSocketInboundGateway;
 import org.springframework.messaging.rsocket.RSocketStrategies;
 
@@ -35,6 +36,18 @@ public class RSocketInboundGatewaySpec extends MessagingGatewaySpec<RSocketInbou
 		super(new RSocketInboundGateway(path));
 	}
 
+	/**
+	 * Configure a set of {@link RSocketInteractionModel} the endpoint is going to be mapped onto.
+	 * @param interactionModels the {@link RSocketInteractionModel}s for mapping.
+	 * @return the spec.
+	 * @since 5.2.2
+	 * @see RSocketInboundGateway#setInteractionModels(RSocketInteractionModel...)
+	 */
+	public RSocketInboundGatewaySpec interactionModels(RSocketInteractionModel... interactionModels) {
+		this.target.setInteractionModels(interactionModels);
+		return this;
+	}
+
 	/**
 	 * Configure {@link RSocketStrategies} instead of a default one.
 	 * @param rsocketStrategies the {@link RSocketStrategies} to use.

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/dsl/RSocketOutboundGatewaySpec.java

@@ -24,6 +24,7 @@
 import org.springframework.integration.expression.FunctionExpression;
 import org.springframework.integration.expression.ValueExpression;
 import org.springframework.integration.rsocket.ClientRSocketConnector;
+import org.springframework.integration.rsocket.RSocketInteractionModel;
 import org.springframework.integration.rsocket.outbound.RSocketOutboundGateway;
 import org.springframework.messaging.Message;
 import org.springframework.util.MimeType;
@@ -62,9 +63,22 @@ public RSocketOutboundGatewaySpec clientRSocketConnector(ClientRSocketConnector
 	 * @param command the {@link RSocketOutboundGateway.Command} to use.
 	 * @return the spec
 	 * @see RSocketOutboundGateway#setCommand(RSocketOutboundGateway.Command)
+	 * @deprecated in favor of {@link #interactionModel(RSocketInteractionModel)}
 	 */
+	@Deprecated
 	public RSocketOutboundGatewaySpec command(RSocketOutboundGateway.Command command) {
-		return command(new ValueExpression<>(command));
+		return interactionModel(new ValueExpression<>(command));
+	}
+
+	/**
+	 * Configure a {@link RSocketInteractionModel} for RSocket request type.
+	 * @param interactionModel the {@link RSocketInteractionModel} to use.
+	 * @return the spec
+	 * @see RSocketOutboundGateway#setInteractionModel(RSocketInteractionModel)
+	 * @since 5.2.2
+	 */
+	public RSocketOutboundGatewaySpec interactionModel(RSocketInteractionModel interactionModel) {
+		return interactionModel(new ValueExpression<>(interactionModel));
 	}
 
 	/**
@@ -73,32 +87,75 @@ public RSocketOutboundGatewaySpec command(RSocketOutboundGateway.Command command
 	 * @param commandFunction the {@code Function} to use.
 	 * @param <P> the expected request message payload type.
 	 * @return the spec
-	 * @see RSocketOutboundGateway#setCommandExpression(Expression)
+	 * @see RSocketOutboundGateway#setInteractionModelExpression(Expression)
+	 * @deprecated in favor of {@link #interactionModel(Function)}
 	 */
+	@Deprecated
 	public <P> RSocketOutboundGatewaySpec command(Function<Message<P>, ?> commandFunction) {
-		return command(new FunctionExpression<>(commandFunction));
+		return interactionModel(commandFunction);
+	}
+
+	/**
+	 * Configure a {@code Function} to evaluate a {@link RSocketInteractionModel}
+	 * for RSocket request type at runtime against a request message.
+	 * @param interactionModelFunction the {@code Function} to use.
+	 * @param <P> the expected request message payload type.
+	 * @return the spec
+	 * @see RSocketOutboundGateway#setInteractionModelExpression(Expression)
+	 * @since 5.2.2
+	 */
+	public <P> RSocketOutboundGatewaySpec interactionModel(Function<Message<P>, ?> interactionModelFunction) {
+		return interactionModel(new FunctionExpression<>(interactionModelFunction));
 	}
 
 	/**
 	 * 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)
+	 * @see RSocketOutboundGateway#setInteractionModelExpression(Expression)
+	 * @deprecated in favor of {@link #interactionModel(String)}
 	 */
+	@Deprecated
 	public RSocketOutboundGatewaySpec command(String commandExpression) {
-		return command(PARSER.parseExpression(commandExpression));
+		return interactionModel(commandExpression);
+	}
+
+	/**
+	 * Configure a SpEL expression to evaluate a {@link RSocketInteractionModel}
+	 * for RSocket request type at runtime against a request message.
+	 * @param interactionModelExpression the SpEL expression to use.
+	 * @return the spec
+	 * @see RSocketOutboundGateway#setInteractionModelExpression(Expression)
+	 * @since 5.2.2
+	 */
+	public RSocketOutboundGatewaySpec interactionModel(String interactionModelExpression) {
+		return interactionModel(PARSER.parseExpression(interactionModelExpression));
 	}
 
 	/**
 	 * 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)
+	 * @see RSocketOutboundGateway#setInteractionModelExpression(Expression)
+	 * @deprecated in favor of {@link #interactionModel(Expression)}
 	 */
+	@Deprecated
 	public RSocketOutboundGatewaySpec command(Expression commandExpression) {
-		this.target.setCommandExpression(commandExpression);
+		return interactionModel(commandExpression);
+	}
+
+	/**
+	 * Configure a SpEL expression to evaluate a {@link RSocketInteractionModel}
+	 * for RSocket request type at runtime against a request message.
+	 * @param interactionModelExpression the SpEL expression to use.
+	 * @return the spec
+	 * @see RSocketOutboundGateway#setInteractionModelExpression(Expression)
+	 * @since 5.2.2
+	 */
+	public RSocketOutboundGatewaySpec interactionModel(Expression interactionModelExpression) {
+		this.target.setInteractionModelExpression(interactionModelExpression);
 		return this;
 	}
 

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

@@ -30,6 +30,7 @@
 import org.springframework.integration.rsocket.AbstractRSocketConnector;
 import org.springframework.integration.rsocket.ClientRSocketConnector;
 import org.springframework.integration.rsocket.IntegrationRSocketEndpoint;
+import org.springframework.integration.rsocket.RSocketInteractionModel;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
@@ -74,6 +75,8 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In
 
 	private final String[] path;
 
+	private RSocketInteractionModel[] interactionModels = RSocketInteractionModel.values();
+
 	private RSocketStrategies rsocketStrategies = RSocketStrategies.create();
 
 	@Nullable
@@ -112,6 +115,21 @@ public void setRSocketConnector(AbstractRSocketConnector rsocketConnector) {
 		this.rsocketConnector = rsocketConnector;
 	}
 
+	/**
+	 * Configure a set of {@link RSocketInteractionModel} this endpoint is mapped onto.
+	 * @param interactionModelsArg the {@link RSocketInteractionModel}s for mapping.
+	 * @since 5.2.2
+	 */
+	public void setInteractionModels(RSocketInteractionModel... interactionModelsArg) {
+		Assert.notNull(interactionModelsArg, "'interactionModelsArg' must not be null");
+		this.interactionModels = Arrays.copyOf(interactionModelsArg, interactionModelsArg.length);
+	}
+
+	@Override
+	public RSocketInteractionModel[] getInteractionModels() {
+		return Arrays.copyOf(this.interactionModels, this.interactionModels.length);
+	}
+
 	/**
 	 * Get an array of the path patterns this endpoint is mapped onto.
 	 * @return the mapping path