INT-3924: Fix RedisQueueInboundGateway

JIRA: https://jira.spring.io/browse/INT-3924

* The `RedisQueueInboundGateway` and `RedisQueueMessageDrivenEndpoint` used `MICROSECONDS` for the wait for stop condition.
  - Change to `MILLISECONDS`
* Both of them may pick up data from the Redis List even after the `stop()`.
  - Add `rightPush()` to the `RedisQueueInboundGateway` to emulate `rollback`.
  - Even if we have `rollback` there we may "steal" messages after the `context.stop()` causing unexpected race condition.
     So change the logic between `stopTimeout` and `receiveTimeout` to `max`, thus we wait for stop more than for `rightPop`.
* Some attributes have been missed for the XML configuration.
  - add `stop-timeout` and `recovery-interval` attributes to the XSD and parsers code.
  - adjust `RedisQueueGatewayIntegrationTests` to use those new attributes.
* Some tests needed polishing:
  - Rework `RedisStoreOutboundChannelAdapterIntegrationTests` to be based on `@ContextConfiguration` for better performance
  - `RedisChannelParserTests` used to `mock` channels for the same Redis topic. Change of of them to different name to avoid `Dispatcher has no subscribers`
    from the `testPubSubChannelUsage()` when the second `SubscribableRedisChannel` doesn't have subscribers.
* Polishing for `redis.adoc`

Address PR comments

* Rework logic to the `Math.min(this.stopTimeout, this.receiveTimeout)` on the adapter/gateway `stop()`
* Fix JavaDocs and `redis.adoc` to describe `stopTimeout` properly.
* Rework `RedisQueueGatewayIntegrationTests` to be based on the `randomUUID` for the queue name to send and receive.

Address PR comments.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-2016 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.
@@ -57,6 +57,8 @@ public class RedisQueueInboundGatewayParser extends AbstractInboundGatewayParser
 		builder.addConstructorArgReference(connectionFactory);
 
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer", true);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "recovery-interval");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2015 the original author or authors
+ * Copyright 2014-2016 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.
@@ -55,7 +55,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
 
 	private static final RedisSerializer<String> stringSerializer = new StringRedisSerializer();
 
-	public static final long DEFAULT_RECEIVE_TIMEOUT = 5000;
+	public static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
 
 	public static final long DEFAULT_RECOVERY_INTERVAL = 5000;
 
@@ -75,14 +75,14 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
 
 	private volatile long recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
 
-	private volatile long stopTimeout = DEFAULT_RECEIVE_TIMEOUT;
-
 	private volatile boolean active;
 
 	private volatile boolean listening;
 
 	private volatile boolean extractPayload = true;
 
+	private volatile Runnable stopCallback;
+
 	/**
 	 * @param queueName         Must not be an empty String
 	 * @param connectionFactory Must not be null
@@ -131,11 +131,12 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
 
 	/**
 	 * @param stopTimeout the timeout to block {@link #doStop()} until the last message
-	 * will be processed or this timeout is reached. Should be less then or equal to
-	 * {@link #receiveTimeout}
+	 * will be processed or this timeout is reached. Should be less than or equal to {@link #receiveTimeout}
+	 * @deprecated since {@literal 4.3} with no-op in favor of delayer call {@code callback.run()}
+	 * in the {@link #stop(Runnable)}.
 	 */
+	@Deprecated
 	public void setStopTimeout(long stopTimeout) {
-		this.stopTimeout = stopTimeout;
 	}
 
 	public void setTaskExecutor(Executor taskExecutor) {
@@ -195,6 +196,10 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
 		}
 		String uuid = null;
 		if (value != null) {
+			if (!active) {
+				this.boundListOperations.rightPush(value);
+				return;
+			}
 			uuid = stringSerializer.deserialize(value);
 			try {
 				value = this.template.boundListOps(uuid).rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
@@ -205,6 +210,11 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
 			}
 			Message<Object> requestMessage = null;
 			if (value != null) {
+				if (!active) {
+					this.template.boundListOps(uuid).rightPush(value);
+					this.boundListOperations.rightPush(stringSerializer.serialize(uuid));
+					return;
+				}
 				if (this.extractPayload) {
 					Object payload = value;
 					if (this.serializer != null) {
@@ -248,6 +258,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
 
 	@Override
 	protected void doStart() {
+		super.doStart();
 		if (!this.active) {
 			this.active = true;
 			this.restart();
@@ -285,23 +296,20 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
 		this.taskExecutor.execute(new ListenerTask());
 	}
 
+	@Override
+	protected void doStop(Runnable callback) {
+		this.stopCallback = callback;
+		doStop();
+	}
+
 	@Override
 	protected void doStop() {
-		try {
-			this.active = false;
-			this.lifecycleCondition.await(Math.min(this.stopTimeout, this.receiveTimeout), TimeUnit.MICROSECONDS);
-		}
-		catch (InterruptedException e) {
-			logger.debug("Thread interrupted while stopping the endpoint");
-			Thread.currentThread().interrupt();
-		}
-		finally {
-			this.listening = false;
-		}
+		super.doStop();
+		this.active = this.listening = false;
 	}
 
 	public boolean isListening() {
-		return listening;
+		return this.listening;
 	}
 
 	/**
@@ -336,21 +344,16 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
 			try {
 				while (RedisQueueInboundGateway.this.active) {
 					RedisQueueInboundGateway.this.listening = true;
-					RedisQueueInboundGateway.this.receiveAndReply();
+					receiveAndReply();
 				}
 			}
 			finally {
 				if (RedisQueueInboundGateway.this.active) {
-					RedisQueueInboundGateway.this.restart();
+					restart();
 				}
-				else {
-					RedisQueueInboundGateway.this.lifecycleLock.lock();
-					try {
-						RedisQueueInboundGateway.this.lifecycleCondition.signalAll();
-					}
-					finally {
-						RedisQueueInboundGateway.this.lifecycleLock.unlock();
-					}
+				else if (RedisQueueInboundGateway.this.stopCallback != null) {
+					RedisQueueInboundGateway.this.stopCallback.run();
+					RedisQueueInboundGateway.this.stopCallback = null;
 				}
 			}
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2015 the original author or authors
+ * Copyright 2013-2016 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.
@@ -13,6 +13,7 @@
  *     See the License for the specific language governing permissions and
  *     limitations under the License.
  */
+
 package org.springframework.integration.redis.inbound;
 
 import java.util.concurrent.Executor;
@@ -73,12 +74,12 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
 
 	private volatile long recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
 
-	private volatile long stopTimeout = DEFAULT_RECEIVE_TIMEOUT;
-
 	private volatile boolean active;
 
 	private volatile boolean listening;
 
+	private volatile Runnable stopCallback;
+
 	/**
 	 * @param queueName         Must not be an empty String
 	 * @param connectionFactory Must not be null
@@ -134,11 +135,13 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
 
 	/**
 	 * @param stopTimeout the timeout to block {@link #doStop()} until the last message will be processed
-	 * or this timeout is reached. Should be less then or equal to {@link #receiveTimeout}
+	 * or this timeout is reached. Should be less than or equal to {@link #receiveTimeout}
 	 * @since 4.0.3
+	 * @deprecated since {@literal 4.3} with no-op in favor of delayer call {@code callback.run()}
+	 * in the {@link #stop(Runnable)}.
 	 */
+	@Deprecated
 	public void setStopTimeout(long stopTimeout) {
-		this.stopTimeout = stopTimeout;
 	}
 
 	public void setTaskExecutor(Executor taskExecutor) {
@@ -268,19 +271,16 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
 		this.taskExecutor.execute(new ListenerTask());
 	}
 
+	@Override
+	protected void doStop(Runnable callback) {
+		this.stopCallback = callback;
+		doStop();
+	}
+
 	@Override
 	protected void doStop() {
-		try {
-			this.active = false;
-			this.lifecycleCondition.await(Math.min(this.stopTimeout, this.receiveTimeout), TimeUnit.MICROSECONDS);
-		}
-		catch (InterruptedException e) {
-			logger.debug("Thread interrupted while stopping the endpoint");
-			Thread.currentThread().interrupt();
-		}
-		finally {
-			this.listening = false;
-		}
+		super.doStop();
+		this.active = this.listening = false;
 	}
 
 	public boolean isListening() {
@@ -327,14 +327,9 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
 				if (RedisQueueMessageDrivenEndpoint.this.active) {
 					RedisQueueMessageDrivenEndpoint.this.restart();
 				}
-				else {
-					RedisQueueMessageDrivenEndpoint.this.lifecycleLock.lock();
-					try {
-						RedisQueueMessageDrivenEndpoint.this.lifecycleCondition.signalAll();
-					}
-					finally {
-						RedisQueueMessageDrivenEndpoint.this.lifecycleLock.unlock();
-					}
+				else if (RedisQueueMessageDrivenEndpoint.this.stopCallback != null) {
+					RedisQueueMessageDrivenEndpoint.this.stopCallback.run();
+					RedisQueueMessageDrivenEndpoint.this.stopCallback = null;
 				}
 			}
 		}

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

@@ -506,6 +506,24 @@
 					</xsd:appinfo>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="receive-timeout" type="xsd:string" default="1000">
+				<xsd:annotation>
+					<xsd:documentation>
+						Specify the timeout in milliseconds to wait for the result of the
+						'rightPop' operation on Redis queue.
+						Default is 1 second.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="recovery-interval" type="xsd:string" default="5000">
+				<xsd:annotation>
+					<xsd:documentation>
+						Specify the time in milliseconds for which the listener task should sleep after catching
+						an Exception on a Redis operation, before restarting the listener task.
+						Default is 5 seconds.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
 			<xsd:attribute name="task-executor" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[

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

@@ -1,16 +1,12 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <beans xmlns="http://www.springframework.org/schema/beans"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xmlns:int="http://www.springframework.org/schema/integration"
-	xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
-	xmlns:redis="http://www.springframework.org/schema/redis"
-	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
-		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
-		http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd
-		http://www.springframework.org/schema/redis http://www.springframework.org/schema/redis/spring-redis.xsd">
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd">
 
-	<int-redis:publish-subscribe-channel id="redisChannel" topic-name="si.test.topic"
-			serializer="redisSerializer"/>
+	<int-redis:publish-subscribe-channel id="redisChannel" topic-name="si.test.topic.parser"
+										 serializer="redisSerializer"/>
 
 	<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
 		<property name="port" value="#{T(org.springframework.integration.redis.rules.RedisAvailableRule).REDIS_PORT}"/>
@@ -19,6 +15,6 @@
 	<bean id="redisSerializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
 
 	<int-redis:publish-subscribe-channel id="redisChannelWithSubLimit" topic-name="si.test.topic"
-			serializer="redisSerializer" max-subscribers="1" />
+										 serializer="redisSerializer" max-subscribers="1"/>
 
 </beans>

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

@@ -56,7 +56,7 @@ public class RedisChannelParserTests extends RedisAvailableTests {
 		RedisSerializer<?> redisSerializer = TestUtils.getPropertyValue(redisChannel, "serializer", RedisSerializer.class);
 		assertEquals(connectionFactory, context.getBean("redisConnectionFactory"));
 		assertEquals(redisSerializer, context.getBean("redisSerializer"));
-		assertEquals("si.test.topic", TestUtils.getPropertyValue(redisChannel, "topicName"));
+		assertEquals("si.test.topic.parser", TestUtils.getPropertyValue(redisChannel, "topicName"));
 		assertEquals(Integer.MAX_VALUE, TestUtils.getPropertyValue(
 				TestUtils.getPropertyValue(redisChannel, "dispatcher"), "maxSubscribers", Integer.class).intValue());
 		redisChannel = context.getBean("redisChannelWithSubLimit", SubscribableChannel.class);

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

@@ -21,10 +21,12 @@
 
 	<int:channel id="requestChannel"/>
 
+	<bean id="redisQueue" class="java.util.UUID" factory-method="randomUUID"/>
+
 	<int-redis:queue-outbound-gateway id="outboundGateway"
 									  request-channel="sendChannel"
-									  queue="foo"
-									  reply-timeout="2000"
+									  queue="#{redisQueue.toString()}"
+									  reply-timeout="1000"
 									  requires-reply="true"
 									  reply-channel="outputChannel"
 									  serializer="serializer"/>
@@ -33,10 +35,11 @@
 	<bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
 
 	<int-redis:queue-inbound-gateway id="inboundGateway"
-									 queue="foo"
+									 queue="#{redisQueue.toString()}"
 									 request-channel="requestChannel"
 									 serializer="serializer"
 									 reply-timeout="20001"
+									 receive-timeout="100"
 									 request-timeout="20000"/>
 
 	<int:service-activator input-channel="requestChannel" expression="payload.toUpperCase()"/>

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-2015 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.
@@ -25,6 +25,9 @@ import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 import org.springframework.integration.channel.DirectChannel;
@@ -33,6 +36,7 @@ import org.springframework.integration.redis.inbound.RedisQueueInboundGateway;
 import org.springframework.integration.redis.outbound.RedisQueueOutboundGateway;
 import org.springframework.integration.redis.rules.RedisAvailable;
 import org.springframework.integration.redis.rules.RedisAvailableTests;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
@@ -41,6 +45,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author David Liu
+ * @author Artem Bilan
  * @since 4.1
  */
 @ContextConfiguration
@@ -48,6 +53,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 @DirtiesContext
 public class RedisQueueGatewayIntegrationTests extends RedisAvailableTests {
 
+	@Value("#{redisQueue.toString().bytes}")
+	private byte[] queueName;
+
 	@Autowired
 	@Qualifier("sendChannel")
 	private DirectChannel sendChannel;
@@ -62,6 +70,11 @@ public class RedisQueueGatewayIntegrationTests extends RedisAvailableTests {
 	@Autowired
 	private RedisQueueOutboundGateway outboundGateway;
 
+	public void setup() {
+		RedisConnectionFactory jcf = getConnectionFactoryForTest();
+		jcf.getConnection().del(this.queueName);
+	}
+
 	@Test
 	@RedisAvailable
 	public void testRequestWithReply() throws Exception {
@@ -74,6 +87,8 @@ public class RedisQueueGatewayIntegrationTests extends RedisAvailableTests {
 	@Test
 	@RedisAvailable
 	public void testInboundGatewayStop() throws Exception {
+		Long receiveTimeout = TestUtils.getPropertyValue(this.inboundGateway, "receiveTimeout", Long.class);
+		this.inboundGateway.setReceiveTimeout(1);
 		this.inboundGateway.stop();
 		try {
 			this.sendChannel.send(new GenericMessage<String>("test1"));
@@ -82,6 +97,7 @@ public class RedisQueueGatewayIntegrationTests extends RedisAvailableTests {
 			assertTrue(e.getMessage().contains("No reply produced"));
 		}
 		finally {
+			this.inboundGateway.setReceiveTimeout(receiveTimeout);
 			this.inboundGateway.start();
 		}
 	}

spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2015 the original author or authors
+ * Copyright 2013-2016 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.
@@ -240,7 +240,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
 		endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
 		endpoint.setOutputChannel(new DirectChannel());
 		endpoint.setReceiveTimeout(1000);
-		endpoint.setStopTimeout(100);
 
 		ExecutorService executorService = Executors.newCachedThreadPool();
 		endpoint.setTaskExecutor(executorService);
@@ -252,11 +251,23 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
 		dfa.setPropertyValue("listening", false);
 
 		redisTemplate.boundListOps(queueName).leftPush("foo");
-		endpoint.stop();
+
+		final CountDownLatch stopLatch = new CountDownLatch(1);
+
+		endpoint.stop(new Runnable() {
+
+			@Override
+			public void run() {
+				stopLatch.countDown();
+			}
+
+		});
 
 		executorService.shutdown();
 		assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS));
 
+		assertTrue(stopLatch.await(10, TimeUnit.SECONDS));
+
 		Mockito.verify(boundListOperations).rightPush(Mockito.any(byte[].class));
 	}
 

spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreOutboundChannelAdapterIntegrationTests-context.xml

@@ -19,11 +19,6 @@
 											  collection-type="LIST"
 											  key="pepboys"/>
 
-	<int-redis:store-outbound-channel-adapter id="listWithKeyProvidedAndChannel"
-											  channel="someChannel"
-											  collection-type="LIST"
-											  key="pepboys"/>
-
 	<int-redis:store-outbound-channel-adapter id="mapToZset"
 											  collection-type="ZSET"
 											  key-expression="'presidents'"/>

spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreOutboundChannelAdapterIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2007-2014 the original author or authors
+ * Copyright 2007-2015 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.
@@ -26,8 +26,15 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
 
+import org.junit.After;
+import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
 
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.RedisTemplate;
@@ -52,207 +59,221 @@ import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandlingException;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Oleg Zhurakousky
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.2
  */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
 public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvailableTests {
 
+	private final StringRedisTemplate redisTemplate = new StringRedisTemplate();
+
+	@Autowired
+	private BeanFactory beanFactory;
+
+	@Autowired
+	@Qualifier("listWithKeyAsHeader")
+	private MessageChannel listWithKeyAsHeaderChannel;
+
+	@Autowired
+	@Qualifier("listWithKeyProvided")
+	private MessageChannel listWithKeyProvidedChannel;
+
+	@Autowired
+	@Qualifier("zset")
+	private MessageChannel zsetChannel;
+
+	@Autowired
+	@Qualifier("mapToZset")
+	private MessageChannel mapToZsetChannel;
+
+	@Autowired
+	@Qualifier("mapToMapA")
+	private MessageChannel mapToMapAChannel;
+
+	@Autowired
+	@Qualifier("mapToMapB")
+	private MessageChannel mapToMapBChannel;
+
+	@Autowired
+	@Qualifier("simpleMap")
+	private MessageChannel simpleMapChannel;
+
+	@Autowired
+	@Qualifier("set")
+	private MessageChannel setChannel;
+
+	@Autowired
+	@Qualifier("setNotParsed")
+	private MessageChannel setNotParsedChannel;
+
+	@Autowired
+	@Qualifier("pojoIntoSet")
+	private MessageChannel pojoIntoSetChannel;
+
+	@Autowired
+	@Qualifier("property")
+	private MessageChannel propertyChannel;
+
+	@Autowired
+	@Qualifier("simpleProperty")
+	private MessageChannel simplePropertyChannel;
+
+	@Before
+	@After
+	public void setup() {
+		RedisConnectionFactory jcf = getConnectionFactoryForTest();
+		this.redisTemplate.setConnectionFactory(jcf);
+		this.redisTemplate.afterPropertiesSet();
+
+		this.redisTemplate.delete("pepboys");
+		this.redisTemplate.delete("foo");
+		this.redisTemplate.delete("bar");
+		this.redisTemplate.delete("presidents");
+	}
+
 	@Test
 	@RedisAvailable
-	public void testListWithKeyAsHeader(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-
-		this.deleteKey(jcf, "pepboys");
-		RedisList<String> redisList =
-				new DefaultRedisList<String>("pepboys", this.initTemplate(jcf, new StringRedisTemplate()));
+	public void testListWithKeyAsHeader() {
+		RedisList<String> redisList = new DefaultRedisList<String>("pepboys", this.redisTemplate);
 		assertEquals(0, redisList.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("listWithKeyAsHeader", MessageChannel.class);
 		List<String> pepboys = new ArrayList<String>();
 		pepboys.add("Manny");
 		pepboys.add("Moe");
 		pepboys.add("Jack");
 		Message<List<String>> message = MessageBuilder.withPayload(pepboys).setHeader(RedisHeaders.KEY, "pepboys").build();
-		redisChannel.send(message);
+		this.listWithKeyAsHeaderChannel.send(message);
 
 		assertEquals(3, redisList.size());
-		this.deleteKey(jcf, "pepboys");
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testListWithKeyAsHeaderSimple(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-
-		StringRedisTemplate redisTemplate = new StringRedisTemplate();
-		RedisList<String> redisList =
-				new DefaultRedisList<String>("foo", this.initTemplate(jcf, redisTemplate));
+	public void testListWithKeyAsHeaderSimple() {
+		redisTemplate.delete("foo");
+		RedisList<String> redisList = new DefaultRedisList<String>("foo", this.redisTemplate);
 		assertEquals(0, redisList.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("listWithKeyAsHeader", MessageChannel.class);
 		Message<String> message = MessageBuilder.withPayload("bar").setHeader("redis_key", "foo").build();
-		redisChannel.send(message);
+		this.listWithKeyAsHeaderChannel.send(message);
 
 		assertEquals(1, redisList.size());
-		redisTemplate.delete("foo");
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testListWithProvidedKey(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.deleteKey(jcf, "pepboys");
-		RedisList<String> redisList =
-				new DefaultRedisList<String>("pepboys", this.initTemplate(jcf, new StringRedisTemplate()));
+	public void testListWithProvidedKey() {
+		RedisList<String> redisList = new DefaultRedisList<String>("pepboys", this.redisTemplate);
 		assertEquals(0, redisList.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("listWithKeyProvided", MessageChannel.class);
 		List<String> pepboys = new ArrayList<String>();
 		pepboys.add("Manny");
 		pepboys.add("Moe");
 		pepboys.add("Jack");
 		Message<List<String>> message = MessageBuilder.withPayload(pepboys).build();
-		redisChannel.send(message);
+		this.listWithKeyProvidedChannel.send(message);
 
 		assertEquals(3, redisList.size());
-		this.deleteKey(jcf, "pepboys");
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testZsetSimplePayloadIncrement(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-
-		StringRedisTemplate redisTemplate = new StringRedisTemplate();
-		RedisZSet<String> redisZSet =
-				new DefaultRedisZSet<String>("foo", this.initTemplate(jcf, redisTemplate));
+	public void testZsetSimplePayloadIncrement() {
+		RedisZSet<String> redisZSet = new DefaultRedisZSet<String>("foo", this.redisTemplate);
 		assertEquals(0, redisZSet.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("zset", MessageChannel.class);
 		Message<String> message = MessageBuilder.withPayload("bar").setHeader(RedisHeaders.KEY, "foo").build();
-		redisChannel.send(message);
+		this.zsetChannel.send(message);
 
 		assertEquals(1, redisZSet.size());
 		assertEquals(Double.valueOf(1), redisZSet.score("bar"));
 
-		redisChannel.send(message);
+		this.zsetChannel.send(message);
 
 		assertEquals(1, redisZSet.size());
 		assertEquals(Double.valueOf(2), redisZSet.score("bar"));
-		redisTemplate.delete("foo");
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testZsetSimplePayloadOverwrite(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-
-		StringRedisTemplate redisTemplate = new StringRedisTemplate();
-		RedisZSet<String> redisZSet =
-				new DefaultRedisZSet<String>("foo", this.initTemplate(jcf, redisTemplate));
+	public void testZsetSimplePayloadOverwrite() {
+		RedisZSet<String> redisZSet = new DefaultRedisZSet<String>("foo", this.redisTemplate);
 		assertEquals(0, redisZSet.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("zset", MessageChannel.class);
 		Message<String> message = MessageBuilder.withPayload("bar")
 				.setHeader(RedisHeaders.KEY, "foo")
 				.setHeader(RedisHeaders.ZSET_INCREMENT_SCORE, false)
 				.build();
-		redisChannel.send(message);
+		this.zsetChannel.send(message);
 
 		assertEquals(1, redisZSet.size());
 		assertEquals(Double.valueOf(1), redisZSet.score("bar"));
 
-		redisChannel.send(message);
+		this.zsetChannel.send(message);
 
 		assertEquals(1, redisZSet.size());
 		assertEquals(Double.valueOf(1), redisZSet.score("bar"));
-		redisTemplate.delete("foo");
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testZsetSimplePayloadIncrementBy2(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-
-		StringRedisTemplate redisTemplate = new StringRedisTemplate();
-		RedisZSet<String> redisZSet =
-				new DefaultRedisZSet<String>("foo", this.initTemplate(jcf, redisTemplate));
+	public void testZsetSimplePayloadIncrementBy2() {
+		RedisZSet<String> redisZSet = new DefaultRedisZSet<String>("foo", this.redisTemplate);
 		assertEquals(0, redisZSet.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("zset", MessageChannel.class);
 		Message<String> message = MessageBuilder.withPayload("bar")
 				.setHeader(RedisHeaders.KEY, "foo")
 				.setHeader(RedisHeaders.ZSET_SCORE, 2)
 				.build();
-		redisChannel.send(message);
+		this.zsetChannel.send(message);
 
 		assertEquals(1, redisZSet.size());
 		assertEquals(Double.valueOf(2), redisZSet.score("bar"));
 
-		redisChannel.send(message);
+		this.zsetChannel.send(message);
 
 		assertEquals(1, redisZSet.size());
 		assertEquals(Double.valueOf(4), redisZSet.score("bar"));
-		redisTemplate.delete("foo");
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testZsetSimplePayloadOverwriteWithHeaderScore(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-
-		StringRedisTemplate redisTemplate = new StringRedisTemplate();
-		RedisZSet<String> redisZSet =
-				new DefaultRedisZSet<String>("foo", this.initTemplate(jcf, redisTemplate));
+	public void testZsetSimplePayloadOverwriteWithHeaderScore() {
+		RedisZSet<String> redisZSet = new DefaultRedisZSet<String>("foo", this.redisTemplate);
 		assertEquals(0, redisZSet.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("zset", MessageChannel.class);
 		Message<String> message = MessageBuilder.withPayload("bar")
 				.setHeader(RedisHeaders.KEY, "foo")
 				.setHeader(RedisHeaders.ZSET_INCREMENT_SCORE, false)
 				.setHeader(RedisHeaders.ZSET_SCORE, 2)
 				.build();
-		redisChannel.send(message);
+		this.zsetChannel.send(message);
 
 		assertEquals(1, redisZSet.size());
 		assertEquals(Double.valueOf(2), redisZSet.score("bar"));
 
-		redisChannel.send(MessageBuilder.fromMessage(message).setHeader(RedisHeaders.ZSET_SCORE, 15).build());
+		this.zsetChannel.send(MessageBuilder.fromMessage(message).setHeader(RedisHeaders.ZSET_SCORE, 15).build());
 
 		assertEquals(1, redisZSet.size());
 		assertEquals(Double.valueOf(15), redisZSet.score("bar"));
-		redisTemplate.delete("foo");
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testMapToZsetWithProvidedKey(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.deletePresidents(jcf);
-		RedisZSet<String> redisZset =
-				new DefaultRedisZSet<String>("presidents", this.initTemplate(jcf, new StringRedisTemplate()));
+	public void testMapToZsetWithProvidedKey() {
+		RedisZSet<String> redisZset = new DefaultRedisZSet<String>("presidents", this.redisTemplate);
 		assertEquals(0, redisZset.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("mapToZset", MessageChannel.class);
 		Map<String, Integer> presidents = new HashMap<String, Integer>();
 		presidents.put("John Adams", 18);
 
@@ -262,19 +283,21 @@ public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvail
 		presidents.put("Zachary Taylor", 19);
 
 		Message<Map<String, Integer>> message = MessageBuilder.withPayload(presidents).build();
-		redisChannel.send(message);
+
+		this.mapToZsetChannel.send(message);
 
 		assertEquals(5, redisZset.size());
 		assertEquals(1, redisZset.rangeByScore(18, 18).size());
 		assertEquals(4, redisZset.rangeByScore(18, 19).size());
 		assertEquals(1, redisZset.rangeByScore(21, 21).size());
 
-		RedisStoreWritingMessageHandler handler = context.getBean("mapToZset.handler",
-				RedisStoreWritingMessageHandler.class);
-		assertEquals("'presidents'", TestUtils.getPropertyValue(handler, "keyExpression", SpelExpression.class).getExpressionString());
+		RedisStoreWritingMessageHandler handler =
+				this.beanFactory.getBean("mapToZset.handler", RedisStoreWritingMessageHandler.class);
+		assertEquals("'presidents'", TestUtils.getPropertyValue(handler, "keyExpression.expression"));
 
 		// test default (increment by score) behavior
-		redisChannel.send(message);
+		this.mapToZsetChannel.send(message);
+
 		assertEquals(5, redisZset.size());
 		assertEquals(1, redisZset.rangeByScore(36, 36).size());
 		assertEquals(4, redisZset.rangeByScore(36, 38).size());
@@ -282,60 +305,51 @@ public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvail
 
 		// test overwrite score behavior
 		presidents.put("Barack Obama", 31);
-		redisChannel.send(MessageBuilder.fromMessage(message).setHeader(RedisHeaders.ZSET_INCREMENT_SCORE, false).build());
+
+		this.mapToZsetChannel.send(MessageBuilder.fromMessage(message)
+				.setHeader(RedisHeaders.ZSET_INCREMENT_SCORE, false)
+				.build());
+
 		assertEquals(5, redisZset.size());
 		assertEquals(1, redisZset.rangeByScore(18, 18).size());
 		assertEquals(4, redisZset.rangeByScore(18, 19).size());
 		assertEquals(1, redisZset.rangeByScore(31, 31).size());
-		this.deletePresidents(jcf);
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testMapToMapWithProvidedKey(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.deleteKey(jcf, "pepboys");
-		RedisMap<String, String> redisMap =
-				new DefaultRedisMap<String, String>("pepboys",
-						this.initTemplate(jcf, new StringRedisTemplate()));
+	public void testMapToMapWithProvidedKey() {
+		RedisMap<String, String> redisMap = new DefaultRedisMap<String, String>("pepboys", this.redisTemplate);
 
 		assertEquals(0, redisMap.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("mapToMapA", MessageChannel.class);
 		Map<String, String> pepboys = new HashMap<String, String>();
 		pepboys.put("1", "Manny");
 		pepboys.put("2", "Moe");
 		pepboys.put("3", "Jack");
 
 		Message<Map<String, String>> message = MessageBuilder.withPayload(pepboys).build();
-		redisChannel.send(message);
+		this.mapToMapAChannel.send(message);
 		assertEquals("Manny", redisMap.get("1"));
 		assertEquals("Moe", redisMap.get("2"));
 		assertEquals("Jack", redisMap.get("3"));
 
-		RedisStoreWritingMessageHandler handler = context.getBean("mapToMapA.handler",
+		RedisStoreWritingMessageHandler handler = this.beanFactory.getBean("mapToMapA.handler",
 				RedisStoreWritingMessageHandler.class);
-		assertEquals("pepboys", TestUtils.getPropertyValue(handler, "keyExpression", LiteralExpression.class).getExpressionString());
-		assertEquals("'foo'", TestUtils.getPropertyValue(handler, "mapKeyExpression", SpelExpression.class).getExpressionString());
-		this.deleteKey(jcf, "pepboys");
-		context.close();
+		assertEquals("pepboys",
+				TestUtils.getPropertyValue(handler, "keyExpression", LiteralExpression.class).getExpressionString());
+		assertEquals("'foo'",
+				TestUtils.getPropertyValue(handler, "mapKeyExpression", SpelExpression.class).getExpressionString());
 	}
 
-	@Test(expected=MessageHandlingException.class) // map key is not provided
+	@Test(expected = MessageHandlingException.class) // map key is not provided
 	@RedisAvailable
-	public void testMapToMapAsSingleEntryWithKeyAsHeaderFail(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.deleteKey(jcf, "pepboys");
+	public void testMapToMapAsSingleEntryWithKeyAsHeaderFail() {
 		RedisMap<String, Map<String, String>> redisMap =
-				new DefaultRedisMap<String, Map<String, String>>("pepboys",
-						this.initTemplate(jcf, new RedisTemplate<String, Map<String, Map<String, String>>>()));
+				new DefaultRedisMap<String, Map<String, String>>("pepboys", this.redisTemplate);
 
 		assertEquals(0, redisMap.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("mapToMapB", MessageChannel.class);
 		Map<String, String> pepboys = new HashMap<String, String>();
 		pepboys.put("1", "Manny");
 		pepboys.put("2", "Moe");
@@ -343,54 +357,47 @@ public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvail
 
 		Message<Map<String, String>> message = MessageBuilder.withPayload(pepboys).
 				setHeader(RedisHeaders.KEY, "pepboys").build();
-		redisChannel.send(message);
-		this.deleteKey(jcf, "pepboys");
-		context.close();
+
+		this.mapToMapBChannel.send(message);
 	}
 
-	@Test(expected=MessageHandlingException.class) // key is not provided
+	@Test(expected = MessageHandlingException.class) // key is not provided
 	@RedisAvailable
-	public void testMapToMapNoKey(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.deleteKey(jcf, "pepboys");
+	public void testMapToMapNoKey() {
 		RedisTemplate<String, Map<String, Map<String, String>>> redisTemplate = new RedisTemplate<String, Map<String, Map<String, String>>>();
 		redisTemplate.setKeySerializer(new StringRedisSerializer());
 		redisTemplate.setHashKeySerializer(new StringRedisSerializer());
+		redisTemplate.setConnectionFactory(getConnectionFactoryForTest());
+		redisTemplate.afterPropertiesSet();
+
 		RedisMap<String, Map<String, String>> redisMap =
-				new DefaultRedisMap<String, Map<String, String>>("pepboys",
-						this.initTemplate(jcf, redisTemplate));
+				new DefaultRedisMap<String, Map<String, String>>("pepboys", redisTemplate);
 
 		assertEquals(0, redisMap.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("mapToMapB", MessageChannel.class);
 		Map<String, String> pepboys = new HashMap<String, String>();
 		pepboys.put("1", "Manny");
 		pepboys.put("2", "Moe");
 		pepboys.put("3", "Jack");
 
 		Message<Map<String, String>> message = MessageBuilder.withPayload(pepboys).build();
-		redisChannel.send(message);
-		this.deleteKey(jcf, "pepboys");
-		context.close();
+		this.mapToMapBChannel.send(message);
 	}
 
 	@Test
 	@RedisAvailable
-	public void testMapToMapAsSingleEntryWithKeyAsHeader(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.deleteKey(jcf, "pepboys");
+	public void testMapToMapAsSingleEntryWithKeyAsHeader() {
 		RedisTemplate<String, Map<String, Map<String, String>>> redisTemplate = new RedisTemplate<String, Map<String, Map<String, String>>>();
 		redisTemplate.setKeySerializer(new StringRedisSerializer());
 		redisTemplate.setHashKeySerializer(new StringRedisSerializer());
+		redisTemplate.setConnectionFactory(getConnectionFactoryForTest());
+		redisTemplate.afterPropertiesSet();
+
 		RedisMap<String, Map<String, String>> redisMap =
-				new DefaultRedisMap<String, Map<String, String>>("pepboys",
-						this.initTemplate(jcf, redisTemplate));
+				new DefaultRedisMap<String, Map<String, String>>("pepboys", redisTemplate);
 
 		assertEquals(0, redisMap.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("mapToMapB", MessageChannel.class);
 		Map<String, String> pepboys = new HashMap<String, String>();
 		pepboys.put("1", "Manny");
 		pepboys.put("2", "Moe");
@@ -398,182 +405,122 @@ public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvail
 
 		Message<Map<String, String>> message = MessageBuilder.withPayload(pepboys).
 				setHeader(RedisHeaders.KEY, "pepboys").setHeader(RedisHeaders.MAP_KEY, "foo").build();
-		redisChannel.send(message);
+		this.mapToMapBChannel.send(message);
 		Map<String, String> pepboyz = redisMap.get("foo");
 
 		assertEquals("Manny", pepboyz.get("1"));
 		assertEquals("Moe", pepboyz.get("2"));
 		assertEquals("Jack", pepboyz.get("3"));
-		this.deleteKey(jcf, "pepboys");
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testStoreSimpleStringInMap(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.deleteKey(jcf, "bar");
-		StringRedisTemplate redisTemplate = new StringRedisTemplate();
-		RedisMap<String, String> redisMap =
-				new DefaultRedisMap<String, String>("bar",
-						this.initTemplate(jcf, redisTemplate));
+	public void testStoreSimpleStringInMap() {
+		RedisMap<String, String> redisMap = new DefaultRedisMap<String, String>("bar", this.redisTemplate);
 
 		assertEquals(0, redisMap.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("simpleMap", MessageChannel.class);
-
 		Message<String> message = MessageBuilder.withPayload("hello, world!").
 				setHeader(RedisHeaders.KEY, "bar").setHeader(RedisHeaders.MAP_KEY, "foo").build();
-		redisChannel.send(message);
+
+		this.simpleMapChannel.send(message);
 		String hello = redisMap.get("foo");
 
 		assertEquals("hello, world!", hello);
-		this.deleteKey(jcf, "bar");
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testSetWithKeyAsHeader(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.deleteKey(jcf, "pepboys");
-		RedisSet<String> redisSet =
-				new DefaultRedisSet<String>("pepboys", this.initTemplate(jcf, new StringRedisTemplate()));
+	public void testSetWithKeyAsHeader() {
+		RedisSet<String> redisSet = new DefaultRedisSet<String>("pepboys", this.redisTemplate);
 		assertEquals(0, redisSet.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("set", MessageChannel.class);
 		Set<String> pepboys = new HashSet<String>();
 		pepboys.add("Manny");
 		pepboys.add("Moe");
 		pepboys.add("Jack");
 		Message<Set<String>> message = MessageBuilder.withPayload(pepboys).setHeader("redis_key", "pepboys").build();
-		redisChannel.send(message);
+		this.setChannel.send(message);
 
 		assertEquals(3, redisSet.size());
-		this.deleteKey(jcf, "pepboys");
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testSetWithKeyAsHeaderSimple(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		StringRedisTemplate redisTemplate = new StringRedisTemplate();
-		RedisSet<String> redisSet =
-				new DefaultRedisSet<String>("foo", this.initTemplate(jcf, redisTemplate));
+	public void testSetWithKeyAsHeaderSimple() {
+		RedisSet<String> redisSet = new DefaultRedisSet<String>("foo", this.redisTemplate);
 		assertEquals(0, redisSet.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("set", MessageChannel.class);
 		Message<String> message = MessageBuilder.withPayload("foo")
 				.setHeader(RedisHeaders.KEY, "foo").build();
-		redisChannel.send(message);
+		this.setChannel.send(message);
 
 		assertEquals(1, redisSet.size());
-		redisTemplate.delete("foo");
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testSetWithKeyAsHeaderNotParsed(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.deleteKey(jcf, "pepboys");
-		RedisTemplate<String, String> redisTemplate = new RedisTemplate<String, String>();
-		redisTemplate.setKeySerializer(new StringRedisSerializer());
-		redisTemplate.setHashKeySerializer(new StringRedisSerializer());
-		RedisSet<String> redisSet =
-				new DefaultRedisSet<String>("pepboys", this.initTemplate(jcf, redisTemplate));
+	public void testSetWithKeyAsHeaderNotParsed() {
+		RedisSet<String> redisSet = new DefaultRedisSet<String>("pepboys", this.redisTemplate);
 		assertEquals(0, redisSet.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("setNotParsed", MessageChannel.class);
 		Set<String> pepboys = new HashSet<String>();
 		pepboys.add("Manny");
 		pepboys.add("Moe");
 		pepboys.add("Jack");
 		Message<Set<String>> message = MessageBuilder.withPayload(pepboys).setHeader("redis_key", "pepboys").build();
-		redisChannel.send(message);
+		this.setNotParsedChannel.send(message);
 
 		assertEquals(1, redisSet.size());
-		this.deleteKey(jcf, "pepboys");
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testPojoIntoSet(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.deleteKey(jcf, "pepboys");
-		RedisSet<String> redisSet =
-				new DefaultRedisSet<String>("pepboys", this.initTemplate(jcf, new StringRedisTemplate()));
+	public void testPojoIntoSet() {
+		RedisSet<String> redisSet = new DefaultRedisSet<String>("pepboys", this.redisTemplate);
 		assertEquals(0, redisSet.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("pojoIntoSet", MessageChannel.class);
 		String pepboy = "Manny";
 		Message<String> message = MessageBuilder.withPayload(pepboy).setHeader("redis_key", "pepboys").build();
-		redisChannel.send(message);
+		this.pojoIntoSetChannel.send(message);
 
 		assertEquals(1, redisSet.size());
-		this.deleteKey(jcf, "pepboys");
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testProperties(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.deleteKey(jcf, "pepboys");
-		RedisProperties redisProperties =
-				new RedisProperties("pepboys", this.initTemplate(jcf, new StringRedisTemplate()));
+	public void testProperties() {
+		RedisProperties redisProperties = new RedisProperties("pepboys", this.redisTemplate);
 
 		assertEquals(0, redisProperties.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("property", MessageChannel.class);
 		Properties pepboys = new Properties();
 		pepboys.put("1", "Manny");
 		pepboys.put("2", "Moe");
 		pepboys.put("3", "Jack");
 
 		Message<Properties> message = MessageBuilder.withPayload(pepboys).build();
-		redisChannel.send(message);
+		this.propertyChannel.send(message);
+
 		assertEquals("Manny", redisProperties.get("1"));
 		assertEquals("Moe", redisProperties.get("2"));
 		assertEquals("Jack", redisProperties.get("3"));
-		this.deleteKey(jcf, "pepboys");
-		context.close();
 	}
 
 	@Test
 	@RedisAvailable
-	public void testPropertiesSimple(){
-		RedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		StringRedisTemplate redisTemplate = new StringRedisTemplate();
-		RedisProperties redisProperties =
-				new RedisProperties("foo", this.initTemplate(jcf, redisTemplate));
+	public void testPropertiesSimple() {
+		RedisProperties redisProperties = new RedisProperties("foo", this.redisTemplate);
 
 		assertEquals(0, redisProperties.size());
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
-		MessageChannel redisChannel = context.getBean("simpleProperty", MessageChannel.class);
 		Message<String> message = MessageBuilder.withPayload("bar")
 				.setHeader(RedisHeaders.KEY, "foo")
 				.setHeader("baz", "qux")
 				.build();
-		redisChannel.send(message);
+		this.simplePropertyChannel.send(message);
 
 		assertEquals("bar", redisProperties.get("qux"));
-		redisTemplate.delete("foo");
-		context.close();
 	}
 
-	private <K,V> RedisTemplate<K,V> initTemplate(RedisConnectionFactory rcf, RedisTemplate<K,V> redisTemplate){
-		redisTemplate.setConnectionFactory(rcf);
-		redisTemplate.afterPropertiesSet();
-		return redisTemplate;
-	}
 }

src/reference/asciidoc/redis.adoc

@@ -24,12 +24,12 @@ To connect to Redis you would use one of the implementations of the `RedisConnec
 ----
 public interface RedisConnectionFactory extends PersistenceExceptionTranslator {
 
-	/**
-	 * Provides a suitable connection for interacting with Redis.
-	 *
-	 * @return connection for interacting with Redis.
-	 */
-	RedisConnection getConnection();
+    /**
+     * Provides a suitable connection for interacting with Redis.
+     *
+     * @return connection for interacting with Redis.
+     */
+    RedisConnection getConnection();
 }
 ----
 
@@ -72,7 +72,7 @@ Or in Spring's XML configuration::
 [source,xml]
 ----
 <bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate">
-	<property name="connectionFactory" ref="redisConnectionFactory"/>
+    <property name="connectionFactory" ref="redisConnectionFactory"/>
 </bean>
 ----
 
@@ -337,7 +337,7 @@ It is recommended that this store is used for backing channels, instead of the g
 </bean>
 
 <int:channel id="somePersistentQueueChannel">
-	<int:queue message-store="redisMessageStore"/>
+    <int:queue message-store="redisMessageStore"/>
 <int:channel>
 ----
 
@@ -530,18 +530,17 @@ Since _Spring Integration 4.0_, the Redis Command Gateway is available to perfor
 [source,xml]
 ----
 <int-redis:outbound-gateway
-		request-channel=""  <1>
-		reply-channel=""  <2>
-		requires-reply=""  <3>
-		reply-timeout=""  <4>
-		connection-factory=""  <5>
-		redis-template=""  <6>
-		arguments-serializer=""  <7>
-		command-expression=""  <8>
-		argument-expressions=""  <9>
-		use-command-variable=""  <10>
-		arguments-strategy="" /> <11>
-
+        request-channel=""  <1>
+        reply-channel=""  <2>
+        requires-reply=""  <3>
+        reply-timeout=""  <4>
+        connection-factory=""  <5>
+        redis-template=""  <6>
+        arguments-serializer=""  <7>
+        command-expression=""  <8>
+        argument-expressions=""  <9>
+        use-command-variable=""  <10>
+        arguments-strategy="" /> <11>
 ----
 
 
@@ -597,8 +596,8 @@ For example to get incremented value from Redis Atomic Number:
 [source,xml]
 ----
 <int-redis:outbound-gateway request-channel="requestChannel"
-	reply-channel="replyChannel"
-	command-expression="'INCR'"/>
+    reply-channel="replyChannel"
+    command-expression="'INCR'"/>
 ----
 
 where the Message `payload` should be a name of `redisCounter`, which may be provided by `org.springframework.data.redis.support.atomic.RedisAtomicInteger` bean definition.
@@ -615,16 +614,15 @@ A different UUID is used for each interaction.
 [source,xml]
 ----
 <int-redis:queue-outbound-gateway
-		request-channel=""  <1>
-		reply-channel=""  <2>
-		requires-reply=""  <3>
-		reply-timeout=""  <4>
-		connection-factory=""  <5>
-		queue=""  <6>
-		order=""  <7>
-		serializer=""  <8>
-		extract-payload=""  <9>
-
+        request-channel=""  <1>
+        reply-channel=""  <2>
+        requires-reply=""  <3>
+        reply-timeout=""  <4>
+        connection-factory=""  <5>
+        queue=""  <6>
+        order=""  <7>
+        serializer=""  <8>
+        extract-payload=""/>  <9>
 ----
 
 
@@ -667,21 +665,21 @@ If this attribute is set to `true`, the `serializer` can't be an empty string be
 === Redis Queue Inbound Gateway
 
 Since _Spring Integration 4.1_, the Redis Queue Inbound Gateway is available to perform request and reply scenarios.
-It pops a _conversation_`UUID` from the provided `queue`, then pops the value from the Redis List with that `UUID` as its key and pushes the reply to the Redis List with a key of `UUID + '.reply'`:
+It pops a _conversation_ `UUID` from the provided `queue`, then pops the value from the Redis List with that `UUID` as its key and pushes the reply to the Redis List with a key of `UUID + '.reply'`:
 [source,xml]
 ----
 <int-redis:queue-inbound-gateway
-		request-channel=""  <1>
-		reply-channel=""  <2>
-		executor=""  <3>
-		reply-timeout=""  <4>
-		connection-factory=""  <5>
-		queue=""  <6>
-		order=""  <7>
-		serializer=""  <8>
-		receive-timeout=""  <9>
-		expect-message=""  <10>
-
+        request-channel=""  <1>
+        reply-channel=""  <2>
+        executor=""  <3>
+        reply-timeout=""  <4>
+        connection-factory=""  <5>
+        queue=""  <6>
+        order=""  <7>
+        serializer=""  <8>
+        receive-timeout=""  <9>
+        expect-message=""  <10>
+        recovery-interval=""/>  <11>
 ----
 
 
@@ -706,7 +704,7 @@ Defaults to `redisConnectionFactory`.
 Mutually exclusive with 'redis-template' attribute.
 
 
-<6> The name of the Redis List for the _conversation_`UUID` s.
+<6> The name of the Redis List for the _conversation_ `UUID` s.
 
 
 <7> The order for this inbound gateway when multiple gateway are registered thereby
@@ -725,6 +723,10 @@ Typically is applied for queue-based limited request-channels.
 <10> Specify if this Endpoint expects data from the Redis queue to contain entire `Message` s.
 If this attribute is set to `true`, the `serializer` can't be an empty string because messages require some form of deserialization (JDK serialization by default).
 
+
+<11> The time in milliseconds for which the listener task should sleep after exceptions on the 'right pop' operation,
+before restarting the listener task.
+
 [[redis-lock-registry]]
 === Redis Lock Registry