INT-2918: Add Redis Outbound Command Gateway

JIRA: https://jira.springsource.org/browse/INT-2918

INT-2918: Improvement according PR comments

INT-2918: Polishing according PR comments

Minor Doc Polishing

INT-2918: Change `notNull` message

Polishing

spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java

@@ -34,5 +34,6 @@ public class RedisNamespaceHandler extends AbstractIntegrationNamespaceHandler {
 		registerBeanDefinitionParser("outbound-channel-adapter", new RedisOutboundChannelAdapterParser());
 		registerBeanDefinitionParser("queue-inbound-channel-adapter", new RedisQueueInboundChannelAdapterParser());
 		registerBeanDefinitionParser("queue-outbound-channel-adapter", new RedisQueueOutboundChannelAdapterParser());
+		registerBeanDefinitionParser("outbound-gateway", new RedisOutboundGatewayParser());
 	}
 }

spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundGatewayParser.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2014 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
+ *
+ *      http://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.redis.config;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.redis.outbound.ExpressionArgumentsStrategy;
+import org.springframework.integration.redis.outbound.RedisOutboundGateway;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the {@code <int-redis:outbound-gateway/>} component.
+ *
+ * @author Artem Bilan
+ * @since 4.0
+ */
+public class RedisOutboundGatewayParser extends AbstractConsumerEndpointParser {
+
+	@Override
+	protected String getInputChannelAttributeName() {
+		return "request-channel";
+	}
+
+	@Override
+	protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisOutboundGateway.class);
+
+		String redisTemplate = element.getAttribute("redis-template");
+		String connectionFactory = element.getAttribute("connection-factory");
+		if (StringUtils.hasText(redisTemplate) && StringUtils.hasText(connectionFactory)) {
+			parserContext.getReaderContext().error("Only one of '" + redisTemplate + "' or '"
+					+ connectionFactory + "' is allowed.", element);
+		}
+		if (StringUtils.hasText(redisTemplate)) {
+			builder.addConstructorArgReference(redisTemplate);
+		}
+		else {
+			if (!StringUtils.hasText(connectionFactory)) {
+				connectionFactory = "redisConnectionFactory";
+			}
+			builder.addConstructorArgReference(connectionFactory);
+		}
+
+		String argumentExpressions = element.getAttribute("argument-expressions");
+		boolean hasArgumentExpressions = StringUtils.hasText(argumentExpressions);
+		String argumentsStrategy = element.getAttribute("arguments-strategy");
+		boolean hasArgumentStrategy = element.hasAttribute("arguments-strategy");
+
+		if (hasArgumentExpressions & hasArgumentStrategy) {
+			parserContext.getReaderContext()
+					.error("'argument-expressions' and 'arguments-strategy' are mutually exclusive.", element);
+		}
+
+		if (hasArgumentExpressions) {
+			BeanDefinitionBuilder argumentsBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionArgumentsStrategy.class)
+					.addConstructorArgValue(argumentExpressions)
+					.addConstructorArgValue(element.getAttribute("use-command-variable"));
+			builder.addPropertyValue("argumentsStrategy", argumentsBuilder.getBeanDefinition());
+		}
+		else if (StringUtils.hasLength(argumentsStrategy)) {
+			builder.addPropertyReference("argumentsStrategy", argumentsStrategy);
+		}
+		else if (hasArgumentStrategy) {
+			builder.addPropertyValue("argumentsStrategy", null);
+		}
+
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "requires-reply");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout", "sendTimeout");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "command-expression");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "arguments-serializer");
+
+		return builder;
+	}
+}

spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ArgumentsStrategy.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2014 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
+ *
+ *      http://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.redis.outbound;
+
+import org.springframework.messaging.Message;
+
+/**
+ * @author Artem Bilan
+ * @since 4.0
+ */
+public interface ArgumentsStrategy {
+
+	Object[] resolve(String command, Message<?> message);
+
+}

spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ExpressionArgumentsStrategy.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2014 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
+ *
+ *      http://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.redis.outbound;
+
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.context.IntegrationContextUtils;
+import org.springframework.integration.expression.IntegrationEvaluationContextAware;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+/**
+ * @author Artem Bilan
+ * @since 4.0
+ */
+public class ExpressionArgumentsStrategy implements ArgumentsStrategy, IntegrationEvaluationContextAware, BeanFactoryAware {
+
+	private static final SpelExpressionParser PARSER = new SpelExpressionParser();
+
+	private final Expression[] argumentExpressions;
+
+	private EvaluationContext evaluationContext;
+
+	private final boolean useCommandVariable;
+
+	private BeanFactory beanFactory;
+
+	public ExpressionArgumentsStrategy(String[] argumentExpressions) {
+		this(argumentExpressions, false);
+	}
+
+	public ExpressionArgumentsStrategy(String[] argumentExpressions, boolean useCommandVariable) {
+		Assert.notNull(argumentExpressions, "'argumentExpressions' must not be null");
+		Assert.noNullElements(argumentExpressions, "'argumentExpressions' cannot have null values.");
+		List<Expression> expressions = new LinkedList<Expression>();
+		for (String argumentExpression : argumentExpressions) {
+			expressions.add(PARSER.parseExpression(argumentExpression));
+		}
+		this.argumentExpressions = expressions.toArray(new Expression[expressions.size()]);
+		this.useCommandVariable = useCommandVariable;
+	}
+
+	@Override
+	public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
+		Assert.notNull(evaluationContext, "'evaluationContext' must not be null");
+		this.evaluationContext = evaluationContext;
+	}
+
+	@Override
+	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+		this.beanFactory = beanFactory;
+	}
+
+	@Override
+	public Object[] resolve(String command, Message<?> message) {
+		EvaluationContext evaluationContext = this.evaluationContext;
+
+		if (this.useCommandVariable) {
+			evaluationContext = IntegrationContextUtils.getEvaluationContext(this.beanFactory);
+			evaluationContext.setVariable("cmd", command);
+		}
+
+		List<Object> arguments = new ArrayList<Object>();
+		for (Expression argumentExpression : this.argumentExpressions) {
+			Object argument = argumentExpression.getValue(evaluationContext, message);
+			if (argument != null) {
+				arguments.add(argument);
+			}
+		}
+		return arguments.toArray();
+	}
+
+}

spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisOutboundGateway.java

@@ -0,0 +1,141 @@
+/*
+ * Copyright 2014 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
+ *
+ *      http://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.redis.outbound;
+
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.GenericToStringSerializer;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.expression.IntegrationEvaluationContextAware;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.redis.support.RedisHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * The Gateway component implementation to perform Redis commands with provided arguments and to return command result.
+ *
+ * @author Artem Bilan
+ * @since 4.0
+ */
+public class RedisOutboundGateway extends AbstractReplyProducingMessageHandler
+		implements IntegrationEvaluationContextAware {
+
+	private static final SpelExpressionParser PARSER = new SpelExpressionParser();
+
+	private final RedisTemplate<?, ?> redisTemplate;
+
+	private EvaluationContext evaluationContext;
+
+	private volatile RedisSerializer<Object> argumentsSerializer = new GenericToStringSerializer<Object>(Object.class);
+
+	private volatile Expression commandExpression = PARSER.parseExpression("headers[" + RedisHeaders.COMMAND + "]");
+
+	private volatile ArgumentsStrategy argumentsStrategy = new PayloadArgumentsStrategy();
+
+	public RedisOutboundGateway(RedisTemplate<?, ?> redisTemplate) {
+		Assert.notNull(redisTemplate, "'redisTemplate' must not be null");
+		this.redisTemplate = redisTemplate;
+	}
+
+	public RedisOutboundGateway(RedisConnectionFactory connectionFactory) {
+		Assert.notNull(connectionFactory, "'connectionFactory' must not be null");
+		this.redisTemplate = new RedisTemplate<Object, Object>();
+		this.redisTemplate.setConnectionFactory(connectionFactory);
+		this.redisTemplate.afterPropertiesSet();
+	}
+
+	@Override
+	public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
+		Assert.notNull(evaluationContext, "'evaluationContext' must not be null");
+		this.evaluationContext = evaluationContext;
+	}
+
+	@SuppressWarnings("unchecked")
+	public void setArgumentsSerializer(RedisSerializer<?> serializer) {
+		Assert.notNull(serializer, "'serializer' must not be null");
+		this.argumentsSerializer = (RedisSerializer<Object>) serializer;
+	}
+
+	public void setCommandExpression(String commandExpression) {
+		Assert.hasText(commandExpression, "'commandExpression' must not be an empty string");
+		this.commandExpression = PARSER.parseExpression(commandExpression);
+	}
+
+	public void setArgumentsStrategy(ArgumentsStrategy argumentsStrategy) {
+		this.argumentsStrategy = argumentsStrategy;
+	}
+
+	@Override
+	protected Object handleRequestMessage(Message<?> requestMessage) {
+		final String command = this.commandExpression.getValue(this.evaluationContext, requestMessage, String.class);
+		Assert.notNull(command, "The 'command' must not evaluate to 'null'.");
+		byte[][] args = null;
+		if (this.argumentsStrategy != null) {
+			Object[] arguments = this.argumentsStrategy.resolve(command, requestMessage);
+			if (!ObjectUtils.isEmpty(arguments)) {
+				args = new byte[arguments.length][];
+
+				for (int i = 0; i < arguments.length; i++) {
+					Object argument = arguments[i];
+					byte[] arg = null;
+					if (argument instanceof byte[]) {
+						arg = (byte[]) argument;
+					}
+					else {
+						arg = this.argumentsSerializer.serialize(argument);
+					}
+					args[i] = arg;
+				}
+			}
+		}
+
+		final byte[][] actualArgs = args;
+
+		return this.redisTemplate.execute(new RedisCallback<Object>() {
+
+			@Override
+			public Object doInRedis(RedisConnection connection) throws DataAccessException {
+				return connection.execute(command, actualArgs);
+			}
+
+		});
+	}
+
+	private class PayloadArgumentsStrategy implements ArgumentsStrategy {
+
+		@Override
+		public Object[] resolve(String command, Message<?> message) {
+			Object payload = message.getPayload();
+			if (payload instanceof Object[]) {
+				return (Object[]) payload;
+			}
+			else {
+				return new Object[]{payload};
+			}
+		}
+
+	}
+
+}

spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java

@@ -1,3 +1,19 @@
+/*
+ * Copyright 2014 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
+ *
+ *      http://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.redis.support;
 
 /**
@@ -6,6 +22,7 @@ package org.springframework.integration.redis.support;
  *
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.2
  */
 public class RedisHeaders {
@@ -20,4 +37,6 @@ public class RedisHeaders {
 
 	public static final String ZSET_INCREMENT_SCORE = PREFIX + "zsetIncrementScore";
 
+	public static final String COMMAND = PREFIX + "command";
+
 }

spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-4.0.xsd

@@ -499,6 +499,163 @@
 		</xsd:complexType>
 	</xsd:element>
 
+	<xsd:element name="outbound-gateway">
+		<xsd:annotation>
+			<xsd:documentation>
+				Defines an outbound Redis Command Sending Gateway.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:choice minOccurs="0" maxOccurs="2">
+				<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
+				<xsd:element name="request-handler-advice-chain" type="integration:adviceChainType" minOccurs="0" maxOccurs="1" />
+			</xsd:choice>
+			<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
+			<xsd:attribute name="connection-factory" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						Reference to a RedisConnectionFactory. If none is provided, the default
+						bean name for the reference will be "redisConnectionFactory".
+						Mutually exclusive with 'redis-template' attribute.
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.data.redis.connection.RedisConnectionFactory"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="redis-template" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						RedisTemplate to be used with this gateway.
+						Mutually exclusive with 'connection-factory' attribute.
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.data.redis.core.RedisTemplate"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="request-channel" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="requires-reply" type="xsd:string" use="optional" default="true">
+				<xsd:annotation>
+					<xsd:documentation>
+						Specify whether this outbound gateway must return a non-null value. This value is
+						'false' by default, otherwise a ReplyRequiredException will be thrown when
+						the underlying service returns a null value.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="reply-channel" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="reply-timeout" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+					Allows you to specify how long this gateway will wait for
+					the reply message to be sent successfully to the reply channel
+					before throwing an exception. This attribute only applies when the
+					channel might block, for example when using a bounded queue channel that
+					is currently full.
+
+					Also, keep in mind that when sending to a DirectChannel, the
+					invocation will occur in the sender's thread. Therefore,
+					the failing of the send operation may be caused by other
+					components further downstream.
+
+					The "reply-timeout" attribute maps to the "sendTimeout" property of the
+					underlying 'MessagingTemplate' instance (org.springframework.integration.core.MessagingTemplate).
+
+					The attribute will default, if not specified, to '-1', meaning that
+					by default, the Gateway will wait indefinitely. The value is
+					specified in milliseconds.
+				]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="order" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+								Specifies the order for invocation when this adapter is connected as a
+								subscriber to a SubscribableChannel.
+							]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="arguments-serializer" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<xsd:documentation>
+							Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer.
+							Used to serialize each command argument to byte[] if necessary.
+						</xsd:documentation>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.data.redis.serializer.RedisSerializer"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="command-expression" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						SpEL expression that returns the command key. Default is the 'redis_command' message header.
+						The command must not be evaluated to 'null'.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="argument-expressions" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						Comma-separated SpEL expressions that will be evaluated as command arguments.
+						Mutually exclusive with the 'arguments-strategy' attribute. If 'use-command-variable' is
+						specified to 'true', the '#cmd' variable will be presented within evaluation context.
+						Argument expressions may evaluate to 'null', to support a variable number of arguments.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="use-command-variable" type="xsd:string" default="false">
+				<xsd:annotation>
+					<xsd:documentation>
+						Specifies, if the evaluated Redis command string will be made available
+						as the '#cmd' variable in the SpEL evaluation context in the
+						org.springframework.integration.redis.outbound.ExpressionArgumentsStrategy
+						when 'argument-expressions' is
+						configured, otherwise this attribute is ignored.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="arguments-strategy" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<xsd:documentation>
+							Reference to an instance of org.springframework.integration.redis.outbound.ArgumentsStrategy.
+							Mutually exclusive with the 'argument-expressions' attribute. By default the 'payload' is used
+							as the command argument(s). To disable any strategy (no arguments) this attribute should be
+							configured as an empty string.
+						</xsd:documentation>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.integration.redis.outbound.ArgumentsStrategy"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+		</xsd:complexType>
+	</xsd:element>
+
 	<xsd:complexType name="redisAdapterType">
 		<xsd:annotation>
 			<xsd:documentation>