INT-4343: Add executor option to the RedisInChA

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

* Add a `Executor` setter named `setTaskExecutor` in
`RedisInboundChannelAdapter` to set the container's task executor.

* Add a `task-executor` attribute to
`<int-redis:inbound-channel-adapter>`.

* Update RedisInboundChannelAdapterParser configuration validation
tests.

* Update redis.adoc and whats-new.adoc.

* Polishing according PR comments
* Improve `RedisQueueGatewayIntegrationTests` performance

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -31,6 +31,8 @@ import org.springframework.util.StringUtils;
  * @author Mark Fisher
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Venil Noronha
+ *
  * @since 2.1
  */
 public class RedisInboundChannelAdapterParser extends AbstractChannelAdapterParser {
@@ -49,6 +51,7 @@ public class RedisInboundChannelAdapterParser extends AbstractChannelAdapterPars
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer", true);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
 
 		return builder.getBeanDefinition();
 	}

spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2007-2016 the original author or authors.
+ * Copyright 2007-2017 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.
@@ -18,6 +18,7 @@ package org.springframework.integration.redis.inbound;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Executor;
 
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
@@ -39,6 +40,8 @@ import org.springframework.util.Assert;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Venil Noronha
+ *
  * @since 2.1
  */
 public class RedisInboundChannelAdapter extends MessageProducerSupport {
@@ -75,6 +78,10 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
 		this.messageConverter = messageConverter;
 	}
 
+	public void setTaskExecutor(Executor taskExecutor) {
+		this.container.setTaskExecutor(taskExecutor);
+	}
+
 	@Override
 	public String getComponentType() {
 		return "redis:inbound-channel-adapter";

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

@@ -193,6 +193,19 @@
 					</xsd:appinfo>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="task-executor" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+						A reference to a Spring TaskExecutor (or standard JDK 1.5+ Executor) for executing
+						Redis listener invokers. Default is a SimpleAsyncTaskExecutor.
+					]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="java.util.concurrent.Executor"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
 		</xsd:complexType>
 	</xsd:element>
 

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml

@@ -9,7 +9,14 @@
 	<int-redis:inbound-channel-adapter
 		id="adapter" topics="foo" topic-patterns="f*, b*" channel="receiveChannel" error-channel="testErrorChannel"
 		message-converter="testConverter"
-		serializer="serializer"/>
+		serializer="serializer"
+		task-executor="executor" />
+
+	<bean id="executor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
+		<property name="corePoolSize" value="5" />
+		<property name="maxPoolSize" value="10" />
+		<property name="queueCapacity" value="25" />
+	</bean>
 
 	<int:channel id="receiveChannel">
 		<int:queue />

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java

@@ -22,6 +22,8 @@ import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 
+import java.util.concurrent.Executor;
+
 import org.hamcrest.Matchers;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -49,6 +51,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  * @author Mark Fisher
  * @author Gary Russell
  * @author Gunnar Hillert
+ * @author Venil Noronha
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
@@ -61,9 +64,13 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests {
 	@Autowired
 	private MessageChannel autoChannel;
 
-	@Autowired @Qualifier("autoChannel.adapter")
+	@Autowired
+	@Qualifier("autoChannel.adapter")
 	private RedisInboundChannelAdapter autoChannelAdapter;
 
+	@Autowired
+	private Executor executor;
+
 	@Test
 	public void validateConfiguration() {
 		RedisInboundChannelAdapter adapter = context.getBean("adapter", RedisInboundChannelAdapter.class);
@@ -76,6 +83,10 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests {
 		assertEquals(converterBean, accessor.getPropertyValue("messageConverter"));
 		assertEquals(context.getBean("serializer"), accessor.getPropertyValue("serializer"));
 
+		Object container = accessor.getPropertyValue("container");
+		DirectFieldAccessor containerAccessor = new DirectFieldAccessor(container);
+		assertSame(this.executor, containerAccessor.getPropertyValue("taskExecutor"));
+
 		Object bean = context.getBean("withoutSerializer.adapter");
 		assertNotNull(bean);
 		assertNull(TestUtils.getPropertyValue(bean, "serializer"));
@@ -109,6 +120,7 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests {
 
 	@SuppressWarnings("unused")
 	private static class TestMessageConverter extends SimpleMessageConverter {
+
 	}
 
 }

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueGatewayIntegrationTests-context.xml

@@ -26,16 +26,15 @@
 	<int-redis:queue-outbound-gateway id="outboundGateway"
 									  request-channel="sendChannel"
 									  queue="#{redisQueue.toString()}"
-									  reply-timeout="10000"
-									  requires-reply="true"
+									  reply-timeout="500"
 									  reply-channel="outputChannel"/>
 
 	<int-redis:queue-inbound-gateway id="inboundGateway"
 									 queue="#{redisQueue.toString()}"
 									 request-channel="requestChannel"
-									 reply-timeout="20001"
+									 reply-timeout="200"
 									 receive-timeout="100"
-									 request-timeout="20000"/>
+									 request-timeout="200"/>
 
 	<int:service-activator input-channel="requestChannel" expression="payload + 1"/>
 

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueGatewayIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-2017 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.
@@ -39,16 +39,15 @@ import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit4.SpringRunner;
 
 /**
  * @author David Liu
  * @author Artem Bilan
+ *
  * @since 4.1
  */
-@ContextConfiguration
-@RunWith(SpringJUnit4ClassRunner.class)
+@RunWith(SpringRunner.class)
 @DirtiesContext
 public class RedisQueueGatewayIntegrationTests extends RedisAvailableTests {
 

src/reference/asciidoc/redis.adoc

@@ -132,7 +132,7 @@ The default is a `SimpleMessageConverter`.
 
 Inbound adapters can subscribe to multiple topic names hence the comma-delimited set of values in the `topics` attribute.
 
-Since _Spring Integration 3.0_, the Inbound Adapter, in addition to the existing `topics` attribute, now has the `topic-patterns` attribute.
+Since _version 3.0_, the Inbound Adapter, in addition to the existing `topics` attribute, now has the `topic-patterns` attribute.
 This attribute contains a comma-delimited set of Redis topic patterns.
 For more information regarding Redis publish/subscribe, see http://redis.io/topics/pubsub[Redis Pub/Sub].
 
@@ -140,6 +140,8 @@ Inbound adapters can use a `RedisSerializer` to deserialize the body of Redis Me
 The `serializer` attribute of the `<int-redis:inbound-channel-adapter>` can be set to an empty string, which results in a `null` value for the `RedisSerializer` property.
 In this case the raw `byte[]` bodies of Redis Messages are provided as the message payloads.
 
+Since _version 5.0_, an `Executor` instance can be provided to the Inbound Adapter via the `task-executor` attribute of the `<int-redis:inbound-channel-adapter>`.
+
 [[redis-outbound-channel-adapter]]
 ==== Redis Outbound Channel Adapter
 

src/reference/asciidoc/whats-new.adoc

@@ -265,6 +265,8 @@ The `RedisStoreWritingMessageHandler` is supplied now with additional String-bas
 The `zsetIncrementExpression` can now be configured on the `RedisStoreWritingMessageHandler`, as well.
 In addition this property has been changed from `true` to `false` since `INCR` option on `ZADD` Redis command is optional.
 
+The `RedisInboundChannelAdapter` can now be supplied with an `Executor` for executing Redis listener invokers.
+
 See <<redis>> for more information.
 
 ==== TCP Changes