Merge pull request #611 from olegz/INT-2733/4/6

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

@@ -64,6 +64,8 @@ public class RedisCollectionsInboundChannelAdapterParser extends AbstractPolling
 
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "collection-type");
 
+		RedisParserUtils.setRedisSerializers(StringUtils.hasText(redisTemplate), element, parserContext, builder);
+
 		String beanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
 				builder.getBeanDefinition(), parserContext.getRegistry());
 		return new RuntimeBeanReference(beanName);

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

@@ -62,10 +62,7 @@ public class RedisCollectionsOutboundChannelAdapterParser extends AbstractOutbou
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "collection-type");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload-elements");
 
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "key-serializer");
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "value-serializer");
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "hash-key-serializer");
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "hash-value-serializer");
+		RedisParserUtils.setRedisSerializers(StringUtils.hasText(redisTemplateRef), element, parserContext, builder);
 
 		if (expressionDef != null){
 			builder.addConstructorArgValue(expressionDef);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -16,13 +16,14 @@
 
 package org.springframework.integration.redis.config;
 
+import org.w3c.dom.Element;
+
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.util.StringUtils;
-import org.w3c.dom.Element;
 
 /**
  * @author Oleg Zhurakousky
@@ -46,6 +47,7 @@ public class RedisInboundChannelAdapterParser extends AbstractChannelAdapterPars
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer");
 		return builder.getBeanDefinition();
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -16,13 +16,14 @@
 
 package org.springframework.integration.redis.config;
 
+import org.w3c.dom.Element;
+
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.util.StringUtils;
-import org.w3c.dom.Element;
 
 /**
  * @author Oleg Zhurakousky
@@ -42,6 +43,7 @@ public class RedisOutboundChannelAdapterParser extends AbstractOutboundChannelAd
 		builder.addConstructorArgReference(connectionFactory);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topic", "defaultTopic");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer");
 		return builder.getBeanDefinition();
 	}
 

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

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2007-2012 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.IntegrationNamespaceUtils;
+
+/**
+ *
+ * @author Oleg Zhurakousky
+ * @since 2.2
+ */
+public class RedisParserUtils {
+
+	public static void setRedisSerializers(boolean redisTemplateSet, Element element, ParserContext parserContext, BeanDefinitionBuilder builder){
+		if (redisTemplateSet){
+			if (element.hasAttribute("key-serializer") ||
+				element.hasAttribute("value-serializer") ||
+				element.hasAttribute("hash-key-serializer") ||
+				element.hasAttribute("hash-value-serializer")){
+
+				parserContext.getReaderContext().error("Serializers can not be provided if this adapter is initialized " +
+						"with an external RedisTemplate. You may set serializers directly on the RedisTemplate", element);
+			}
+		}
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "key-serializer");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "value-serializer");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "hash-key-serializer");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "hash-value-serializer");
+	}
+}

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

@@ -23,6 +23,7 @@ import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.listener.ChannelTopic;
 import org.springframework.data.redis.listener.RedisMessageListenerContainer;
 import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 import org.springframework.integration.Message;
 import org.springframework.integration.endpoint.MessageProducerSupport;
@@ -38,27 +39,33 @@ import org.springframework.util.Assert;
 public class RedisInboundChannelAdapter extends MessageProducerSupport {
 
 	private final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
-	
+
 	private volatile MessageConverter messageConverter = new SimpleMessageConverter();
 
 	private volatile String[] topics;
 
+	private volatile RedisSerializer<?> serializer = new StringRedisSerializer();
 
 	public RedisInboundChannelAdapter(RedisConnectionFactory connectionFactory) {
 		Assert.notNull(connectionFactory, "connectionFactory must not be null");
 		this.container.setConnectionFactory(connectionFactory);
 	}
 
+	public void setSerializer(RedisSerializer<?> serializer) {
+		Assert.notNull(serializer, "'serializer' must not be null");
+		this.serializer = serializer;
+	}
 
 	public void setTopics(String... topics) {
 		this.topics = topics;
 	}
-	
+
 	public void setMessageConverter(MessageConverter messageConverter) {
 		Assert.notNull(messageConverter, "messageConverter must not be null");
 		this.messageConverter = messageConverter;
 	}
 
+	@Override
 	public String getComponentType() {
 		return "redis:inbound-channel-adapter";
 	}
@@ -69,23 +76,23 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
 		Assert.notEmpty(this.topics, "at least one topis is required for subscription");
 		MessageListenerDelegate delegate = new MessageListenerDelegate();
 		MessageListenerAdapter adapter = new MessageListenerAdapter(delegate);
-		adapter.setSerializer(new StringRedisSerializer());
+		adapter.setSerializer(this.serializer);
 		List<ChannelTopic> topicList = new ArrayList<ChannelTopic>();
 		for (String topic : this.topics) {
 			topicList.add(new ChannelTopic(topic));
-		}	
+		}
 		adapter.afterPropertiesSet();
 		this.container.addMessageListener(adapter, topicList);
 		this.container.afterPropertiesSet();
 	}
-	
+
 	@Override
 	protected void doStart() {
 		super.doStart();
 		this.container.start();
 	}
 
-	
+
 	@Override
 	protected void doStop() {
 		super.doStop();

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

@@ -21,6 +21,7 @@ import java.util.Collection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
+import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 import org.springframework.data.redis.support.collections.RedisCollectionFactoryBean;
 import org.springframework.data.redis.support.collections.RedisCollectionFactoryBean.CollectionType;
@@ -57,6 +58,16 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport
 
 	private final RedisTemplate<String, ?> redisTemplate;
 
+	private volatile RedisSerializer<?> keySerializer;
+
+	private volatile RedisSerializer<?> valueSerializer;
+
+	private volatile RedisSerializer<?> hashKeySerializer;
+
+	private volatile RedisSerializer<?> hashValueSerializer;
+
+	private volatile boolean usingDefaultTemplate = true;
+
 	/**
 	 * Creates this instance with provided {@link RedisTemplate} and SpEL expression
 	 * which should resolve to a 'key' name of the collection to be used.
@@ -73,6 +84,7 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport
 		Assert.notNull(redisTemplate, "'redisTemplate' must not be null");
 
 		this.redisTemplate = redisTemplate;
+		this.usingDefaultTemplate = false;
 		this.keyExpression = keyExpression;
 	}
 
@@ -96,15 +108,35 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport
 
 		RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
 		redisTemplate.setConnectionFactory(connectionFactory);
-		redisTemplate.setKeySerializer(new StringRedisSerializer());
-		redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
-		redisTemplate.setHashKeySerializer(new JdkSerializationRedisSerializer());
-		redisTemplate.setHashValueSerializer(new JdkSerializationRedisSerializer());
 
 		this.redisTemplate = redisTemplate;
 		this.keyExpression = keyExpression;
 	}
 
+	public void setKeySerializer(RedisSerializer<?> keySerializer) {
+		Assert.state(this.usingDefaultTemplate, "'keySerializer' can not be set if a RedisTemplate was explicitly provided");
+		Assert.notNull(keySerializer, "'keySerializer' must not be null");
+		this.keySerializer = keySerializer;
+	}
+
+	public void setValueSerializer(RedisSerializer<?> valueSerializer) {
+		Assert.state(this.usingDefaultTemplate, "'valueSerializer' can not be set if a RedisTemplate was explicitly provided");
+		Assert.notNull(valueSerializer, "'valueSerializer' must not be null");
+		this.valueSerializer = valueSerializer;
+	}
+
+	public void setHashKeySerializer(RedisSerializer<?> hashKeySerializer) {
+		Assert.state(this.usingDefaultTemplate, "'hashKeySerializer' can not be set if a RedisTemplate was explicitly provided");
+		Assert.notNull(hashKeySerializer, "'hashKeySerializer' must not be null");
+		this.hashKeySerializer = hashKeySerializer;
+	}
+
+	public void setHashValueSerializer(RedisSerializer<?> hashValueSerializer) {
+		Assert.state(this.usingDefaultTemplate, "'hashValueSerializer' can not be set if a RedisTemplate was explicitly provided");
+		Assert.notNull(hashValueSerializer, "'hashValueSerializer' must not be null");
+		this.hashValueSerializer = hashValueSerializer;
+	}
+
 	public void setCollectionType(CollectionType collectionType) {
 		this.collectionType = collectionType;
 	}
@@ -152,6 +184,22 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport
 		else {
 			this.evaluationContext = ExpressionUtils.createStandardEvaluationContext();
 		}
+
+		if (this.usingDefaultTemplate){
+			if (this.keySerializer != null){
+				redisTemplate.setKeySerializer(this.keySerializer);
+			}
+			if (this.valueSerializer != null){
+				redisTemplate.setValueSerializer(this.valueSerializer);
+			}
+			if (this.hashKeySerializer != null){
+				redisTemplate.setHashKeySerializer(this.hashKeySerializer);
+			}
+			if (this.hashValueSerializer != null){
+				redisTemplate.setHashValueSerializer(this.hashValueSerializer);
+			}
+			this.redisTemplate.afterPropertiesSet();
+		}
 	}
 
 	public RedisStore getResource() {

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

@@ -21,7 +21,6 @@ import java.util.Properties;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.BoundSetOperations;
@@ -95,6 +94,8 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand
 
 	private volatile RedisSerializer<?> hashValueSerializer;
 
+	private volatile boolean usingDefaultTemplate = true;
+
 	/**
 	 * Will construct this instance using fully created and initialized instance of
 	 * provided {@link RedisTemplate}
@@ -120,6 +121,7 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand
 	public RedisCollectionPopulatingMessageHandler(RedisTemplate<String, ?> redisTemplate, Expression keyExpression) {
 		Assert.notNull(redisTemplate, "'redisTemplate' must not be null");
 		this.redisTemplate = redisTemplate;
+		this.usingDefaultTemplate = false;
 		if (keyExpression != null) {
 			this.keyExpression = keyExpression;
 		}
@@ -162,21 +164,25 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand
 	}
 
 	public void setKeySerializer(RedisSerializer<?> keySerializer) {
+		Assert.state(this.usingDefaultTemplate, "'keySerializer' can not be set if a RedisTemplate was explicitly provided");
 		Assert.notNull(keySerializer, "'keySerializer' must not be null");
 		this.keySerializer = keySerializer;
 	}
 
 	public void setValueSerializer(RedisSerializer<?> valueSerializer) {
+		Assert.state(this.usingDefaultTemplate, "'valueSerializer' can not be set if a RedisTemplate was explicitly provided");
 		Assert.notNull(valueSerializer, "'valueSerializer' must not be null");
 		this.valueSerializer = valueSerializer;
 	}
 
 	public void setHashKeySerializer(RedisSerializer<?> hashKeySerializer) {
+		Assert.state(this.usingDefaultTemplate, "'hashKeySerializer' can not be set if a RedisTemplate was explicitly provided");
 		Assert.notNull(hashKeySerializer, "'hashKeySerializer' must not be null");
 		this.hashKeySerializer = hashKeySerializer;
 	}
 
 	public void setHashValueSerializer(RedisSerializer<?> hashValueSerializer) {
+		Assert.state(this.usingDefaultTemplate, "'hashValueSerializer' can not be set if a RedisTemplate was explicitly provided");
 		Assert.notNull(hashValueSerializer, "'hashValueSerializer' must not be null");
 		this.hashValueSerializer = hashValueSerializer;
 	}
@@ -234,19 +240,22 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand
 		Assert.state(!this.mapKeyExpressionExplicitlySet ||
 				(this.collectionType == CollectionType.MAP || this.collectionType == CollectionType.PROPERTIES),
 				"'mapKeyExpression' can only be set for CollectionType.MAP or CollectionType.PROPERTIES");
-		if (this.keySerializer != null){
-			redisTemplate.setKeySerializer(this.keySerializer);
+
+		if (this.usingDefaultTemplate){
+			if (this.keySerializer != null){
+				redisTemplate.setKeySerializer(this.keySerializer);
+			}
+			if (this.valueSerializer != null){
+				redisTemplate.setValueSerializer(this.valueSerializer);
+			}
+			if (this.hashKeySerializer != null){
+				redisTemplate.setHashKeySerializer(this.hashKeySerializer);
+			}
+			if (this.hashValueSerializer != null){
+				redisTemplate.setHashValueSerializer(this.hashValueSerializer);
+			}
+			this.redisTemplate.afterPropertiesSet();
 		}
-		if (this.valueSerializer != null){
-			redisTemplate.setValueSerializer(this.valueSerializer);
-		}
-		if (this.hashKeySerializer != null){
-			redisTemplate.setHashKeySerializer(this.hashKeySerializer);
-		}
-		if (this.hashValueSerializer != null){
-			redisTemplate.setHashValueSerializer(this.hashValueSerializer);
-		}
-		this.redisTemplate.afterPropertiesSet();
 	}
 
 	/**

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

@@ -18,9 +18,10 @@ package org.springframework.integration.redis.outbound;
 
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
 import org.springframework.integration.Message;
-import org.springframework.integration.MessagingException;
-import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.support.converter.MessageConverter;
 import org.springframework.integration.support.converter.SimpleMessageConverter;
 import org.springframework.util.Assert;
@@ -29,7 +30,7 @@ import org.springframework.util.Assert;
  * @author Mark Fisher
  * @since 2.1
  */
-public class RedisPublishingMessageHandler implements MessageHandler {
+public class RedisPublishingMessageHandler extends AbstractMessageHandler {
 
 	private final StringRedisTemplate template;
 
@@ -37,12 +38,17 @@ public class RedisPublishingMessageHandler implements MessageHandler {
 
 	private volatile String defaultTopic;
 
+	private volatile RedisSerializer<?> serializer = new StringRedisSerializer();
 
 	public RedisPublishingMessageHandler(RedisConnectionFactory connectionFactory) {
 		Assert.notNull(connectionFactory, "connectionFactory must not be null");
 		this.template = new StringRedisTemplate(connectionFactory);
 	}
 
+	public void setSerializer(RedisSerializer<?> serializer) {
+		Assert.notNull(serializer, "'serializer' must not be null");
+		this.serializer = serializer;
+	}
 
 	public void setMessageConverter(MessageConverter messageConverter) {
 		Assert.notNull(messageConverter, "messageConverter must not be null");
@@ -53,12 +59,6 @@ public class RedisPublishingMessageHandler implements MessageHandler {
 		this.defaultTopic = defaultTopic;
 	}
 
-	public void handleMessage(Message<?> message) throws MessagingException {
-		String topic = this.determineTopic(message);
-		Object value = this.messageConverter.fromMessage(message);
-		this.template.convertAndSend(topic, value.toString());
-	}
-
 	private String determineTopic(Message<?> message) {
 		// TODO: add support for determining topic by evaluating SpEL against the Message
 		Assert.hasText(this.defaultTopic, "Failed to determine Redis topic " +
@@ -66,4 +66,16 @@ public class RedisPublishingMessageHandler implements MessageHandler {
 		return this.defaultTopic;
 	}
 
+	@Override
+	protected void handleMessageInternal(Message<?> message) throws Exception {
+		String topic = this.determineTopic(message);
+		Object value = this.messageConverter.fromMessage(message);
+		this.template.convertAndSend(topic, value.toString());
+	}
+
+	@Override
+	protected void onInit() throws Exception {
+		this.template.setValueSerializer(this.serializer);
+		this.template.afterPropertiesSet();
+	}
 }

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

@@ -191,6 +191,18 @@
 					]]></xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="serializer" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<xsd:documentation>
+						Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer 
+						</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:complexType>
 	</xsd:element>
 
@@ -237,6 +249,18 @@
 							]]></xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="serializer" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<xsd:documentation>
+								Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer 
+								</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:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -394,6 +418,62 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="key-serializer" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<xsd:documentation>
+								Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer
+								to serialize the 'key' value for this collection. Please refer to the JavaDoc of the RedisTemplate 
+								for more detail.
+								</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="value-serializer" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<xsd:documentation>
+								Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer 
+								to serialize the 'value' entered into the collection. Please refer to the JavaDoc of the RedisTemplate 
+								for more detail.
+								</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="hash-key-serializer" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<xsd:documentation>
+								Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer 
+								to be used when serializing hash keys (only relevant for hash-typed collections). 
+								Please refer to the JavaDoc of the RedisTemplate for more detail.
+								</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="hash-value-serializer" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<xsd:documentation>
+								Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer 
+								to be used when serializing hash values (only relevant for hash-typed collections). 
+								Please refer to the JavaDoc of the RedisTemplate for more detail.
+								</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="error-channel" type="xsd:string">
 						<xsd:annotation>
 							<xsd:appinfo>

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

@@ -0,0 +1,237 @@
+/*
+ * Copyright 2002-2012 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+import org.junit.Test;
+
+import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.support.collections.RedisList;
+import org.springframework.data.redis.support.collections.RedisZSet;
+import org.springframework.integration.Message;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.redis.rules.RedisAvailable;
+import org.springframework.integration.redis.rules.RedisAvailableTests;
+
+/**
+ * @author Oleg Zhurakousky
+ * @since 2.2
+ */
+public class RedisCollectionsInboundChannelAdapterIntegrationTests extends RedisAvailableTests{
+
+	@Test
+	@RedisAvailable
+	@SuppressWarnings("unchecked")
+	public void testListInboundConfiguration() throws Exception{
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		this.prepareList(jcf);
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass());
+		SourcePollingChannelAdapter spca = context.getBean("listAdapter", SourcePollingChannelAdapter.class);
+		spca.start();
+		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
+
+
+		Message<RedisList<Object>> message = (Message<RedisList<Object>>) redisChannel.receive(1000);
+		assertNotNull(message);
+		assertEquals(13, message.getPayload().size());
+
+		//poll again, should get the same stuff
+		message = (Message<RedisList<Object>>) redisChannel.receive(1000);
+		assertNotNull(message);
+		assertEquals(13, message.getPayload().size());
+		context.close();
+	}
+
+	@Test
+	@RedisAvailable
+	@SuppressWarnings("unchecked")
+	public void testListInboundConfigurationWithSynchronization() throws Exception{
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		this.prepareList(jcf);
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass());
+		SourcePollingChannelAdapter spca = context.getBean("listAdapterWithSynchronization", SourcePollingChannelAdapter.class);
+		spca.start();
+		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
+
+
+		Message<RedisList<Object>> message = (Message<RedisList<Object>>) redisChannel.receive(1000);
+		assertNotNull(message);
+		assertEquals(13, message.getPayload().size());
+
+		//poll again, should get nothing since the collection was removed during synchronization
+		message = (Message<RedisList<Object>>) redisChannel.receive(1000);
+		assertNull(message);
+
+		spca.stop();
+		context.close();
+	}
+
+	@Test
+	@RedisAvailable
+	@SuppressWarnings("unchecked")
+	public void testListInboundConfigurationWithSynchronizationAndTemplate() throws Exception{
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		this.prepareList(jcf);
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass());
+		SourcePollingChannelAdapter spca = context.getBean("listAdapterWithSynchronizationAndRedisTemplate", SourcePollingChannelAdapter.class);
+		spca.start();
+		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
+
+		Message<RedisList<Object>> message = (Message<RedisList<Object>>) redisChannel.receive(1000);
+		assertNotNull(message);
+		assertEquals(13, message.getPayload().size());
+
+		//poll again, should get nothing since the collection was removed during synchronization
+		message = (Message<RedisList<Object>>) redisChannel.receive(1000);
+		assertNull(message);
+
+		spca.stop();
+		context.close();
+	}
+
+	@Test
+	@RedisAvailable
+	@SuppressWarnings("unchecked")
+	public void testZsetInboundConfiguration(){
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		this.prepareZset(jcf);
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("zset-inbound-adapter.xml", this.getClass());
+		SourcePollingChannelAdapter zsetAdapterNoScore =
+				context.getBean("zsetAdapterNoScore", SourcePollingChannelAdapter.class);
+		zsetAdapterNoScore.start();
+
+		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
+
+		Message<RedisZSet<Object>> message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
+		assertNotNull(message);
+		assertEquals(13, message.getPayload().size());
+
+		//poll again, should get the same stuff
+		message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
+		assertNotNull(message);
+		assertEquals(13, message.getPayload().size());
+
+		zsetAdapterNoScore.stop();
+		context.close();
+	}
+
+	@Test
+	@RedisAvailable
+	@SuppressWarnings("unchecked")
+	public void testZsetInboundConfigurationWithScoreRange(){
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		this.prepareZset(jcf);
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("zset-inbound-adapter.xml", this.getClass());
+		SourcePollingChannelAdapter zsetAdapterWithScoreRange =
+				context.getBean("zsetAdapterWithScoreRange", SourcePollingChannelAdapter.class);
+		zsetAdapterWithScoreRange.start();
+
+		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
+
+		Message<RedisZSet<Object>> message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
+		assertNotNull(message);
+		assertEquals(11, message.getPayload().rangeByScore(18, 20).size());
+
+		//poll again, should get the same stuff
+		message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
+		assertNotNull(message);
+		assertEquals(11, message.getPayload().rangeByScore(18, 20).size());
+
+		zsetAdapterWithScoreRange.stop();
+		context.close();
+	}
+
+	@Test
+	@RedisAvailable
+	@SuppressWarnings("unchecked")
+	public void testZsetInboundConfigurationWithSingleScore(){
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		this.prepareZset(jcf);
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("zset-inbound-adapter.xml", this.getClass());
+		SourcePollingChannelAdapter zsetAdapterWithSingleScore =
+				context.getBean("zsetAdapterWithSingleScore", SourcePollingChannelAdapter.class);
+		zsetAdapterWithSingleScore.start();
+
+		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
+
+		Message<RedisZSet<Object>> message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
+		assertNotNull(message);
+		assertEquals(2, message.getPayload().rangeByScore(18, 18).size());
+
+		//poll again, should get the same stuff
+		message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
+		assertNotNull(message);
+		assertEquals(2, message.getPayload().rangeByScore(18, 18).size());
+
+		zsetAdapterWithSingleScore.stop();
+		context.close();
+	}
+
+	@Test
+	@RedisAvailable
+	@SuppressWarnings("unchecked")
+	public void testZsetInboundConfigurationWithSingleScoreAndSynchronization() throws Exception{
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		this.prepareZset(jcf);
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("zset-inbound-adapter.xml", this.getClass());
+		SourcePollingChannelAdapter zsetAdapterWithSingleScoreAndSynchronization =
+				context.getBean("zsetAdapterWithSingleScoreAndSynchronization", SourcePollingChannelAdapter.class);
+
+		SourcePollingChannelAdapter zsetAdapterNoScore =
+				context.getBean("zsetAdapterNoScore", SourcePollingChannelAdapter.class);
+
+		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
+		QueueChannel otherRedisChannel = context.getBean("otherRedisChannel", QueueChannel.class);
+		// get all 13 presidents
+		zsetAdapterNoScore.start();
+
+		Message<RedisZSet<Object>> message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
+		assertNotNull(message);
+		assertEquals(13, message.getPayload().size());
+
+		zsetAdapterNoScore.stop();
+		Thread.sleep(1000);
+		// get only presidents for 18th century
+		zsetAdapterWithSingleScoreAndSynchronization.start();
+
+		message = (Message<RedisZSet<Object>>) otherRedisChannel.receive(1000);
+		assertNotNull(message);
+		assertEquals(2, message.getPayload().rangeByScore(18, 18).size());
+		//message.getPayload().remo
+		zsetAdapterWithSingleScoreAndSynchronization.stop();
+		Thread.sleep(1000);
+
+		// ... however other elements are still available 13-2=11
+		zsetAdapterNoScore.start();
+		message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
+		assertNotNull(message);
+		assertEquals(11, message.getPayload().size());
+
+		zsetAdapterNoScore.stop();
+		context.close();
+	}
+
+	@Test(expected=BeanDefinitionParsingException.class)
+	public void testTemplateAndCfMutualExclusivity(){
+		new ClassPathXmlApplicationContext("inbound-template-cf-fail.xml", this.getClass());
+	}
+}

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

@@ -16,288 +16,36 @@
 package org.springframework.integration.redis.config;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
 
 import org.junit.Test;
 
 import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
-import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
-import org.springframework.data.redis.core.BoundListOperations;
-import org.springframework.data.redis.core.RedisTemplate;
-import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
-import org.springframework.data.redis.serializer.StringRedisSerializer;
-import org.springframework.data.redis.support.collections.RedisList;
-import org.springframework.data.redis.support.collections.RedisZSet;
-import org.springframework.integration.Message;
-import org.springframework.integration.MessagingException;
-import org.springframework.integration.channel.QueueChannel;
-import org.springframework.integration.core.MessageHandler;
-import org.springframework.integration.core.SubscribableChannel;
-import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
-import org.springframework.integration.redis.rules.RedisAvailable;
-import org.springframework.integration.redis.rules.RedisAvailableTests;
+import org.springframework.data.redis.support.collections.RedisCollectionFactoryBean.CollectionType;
+import org.springframework.expression.spel.standard.SpelExpression;
+import org.springframework.integration.redis.inbound.RedisStoreMessageSource;
+import org.springframework.integration.test.util.TestUtils;
 
 /**
  * @author Oleg Zhurakousky
  * @since 2.2
  */
-public class RedisCollectionsInboundChannelAdapterParserTests extends RedisAvailableTests{
-
+public class RedisCollectionsInboundChannelAdapterParserTests {
 	@Test
-	@RedisAvailable
-	@SuppressWarnings("unchecked")
-	public void testListInboundConfiguration() throws Exception{
-		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.prepareList(jcf);
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass());
-		SourcePollingChannelAdapter spca = context.getBean("listAdapter", SourcePollingChannelAdapter.class);
-		spca.start();
-		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
-
-
-		Message<RedisList<Object>> message = (Message<RedisList<Object>>) redisChannel.receive(1000);
-		assertNotNull(message);
-		assertEquals(13, message.getPayload().size());
-
-		//poll again, should get the same stuff
-		message = (Message<RedisList<Object>>) redisChannel.receive(1000);
-		assertNotNull(message);
-		assertEquals(13, message.getPayload().size());
-		context.close();
-	}
-
-	@Test
-	@RedisAvailable
-	@SuppressWarnings("unchecked")
-	public void testListInboundConfigurationWithSynchronization() throws Exception{
-		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.prepareList(jcf);
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass());
-		SourcePollingChannelAdapter spca = context.getBean("listAdapterWithSynchronization", SourcePollingChannelAdapter.class);
-		spca.start();
-		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
-
-
-		Message<RedisList<Object>> message = (Message<RedisList<Object>>) redisChannel.receive(1000);
-		assertNotNull(message);
-		assertEquals(13, message.getPayload().size());
-
-		//poll again, should get nothing since the collection was removed during synchronization
-		message = (Message<RedisList<Object>>) redisChannel.receive(1000);
-		assertNull(message);
-
-		spca.stop();
-		context.close();
-	}
-
-	@Test
-	@RedisAvailable
-	public void testListInboundConfigurationWithSynchronizationAndRollback() throws Exception{
-		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.prepareList(jcf);
-
-		RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<Object, Object>();
-		redisTemplate.setConnectionFactory(jcf);
-		redisTemplate.setKeySerializer(new StringRedisSerializer());
-		redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
-
-		BoundListOperations<Object, Object> ops = redisTemplate.boundListOps("baz");
-
-		assertTrue(ops.size() == 0);
-
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass());
-		SourcePollingChannelAdapter spca = context.getBean("listAdapterWithSynchronizationAndRollback", SourcePollingChannelAdapter.class);
-		SubscribableChannel redisFailChannel = context.getBean("redisFailChannel", SubscribableChannel.class);
-		redisFailChannel.subscribe(new MessageHandler() {
-
-			public void handleMessage(Message<?> message) throws MessagingException {
-				throw new MessagingException("intentional");
-			}
-		});
-		spca.start();
-
-		Thread.sleep(1000);
-		ops = redisTemplate.boundListOps("baz");
-		assertTrue(ops.size() == 13);
-		ops = redisTemplate.boundListOps("bar");
-		assertTrue(ops.size() == 0);
-
-		spca.stop();
-		context.close();
-	}
-
-	@Test
-	@RedisAvailable
-	@SuppressWarnings("unchecked")
-	public void testListInboundConfigurationWithSynchronizationAndTemplate() throws Exception{
-		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.prepareList(jcf);
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass());
-		SourcePollingChannelAdapter spca = context.getBean("listAdapterWithSynchronizationAndRedisTemplate", SourcePollingChannelAdapter.class);
-		spca.start();
-		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
-
-
-		Message<RedisList<Object>> message = (Message<RedisList<Object>>) redisChannel.receive(1000);
-		assertNotNull(message);
-		assertEquals(13, message.getPayload().size());
-
-		//poll again, should get nothing since the collection was removed during synchronization
-		message = (Message<RedisList<Object>>) redisChannel.receive(1000);
-		assertNull(message);
-
-		spca.stop();
-		context.close();
-	}
-
-	@Test
-	@RedisAvailable
-	@SuppressWarnings("unchecked")
-	public void testListInboundConfigurationWithBeforeCommit() throws Exception{
-		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.prepareList(jcf);
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass());
-		SourcePollingChannelAdapter spca = context.getBean("listAdapterWithSynchronizationBeforeCommit", SourcePollingChannelAdapter.class);
-		spca.start();
-		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
-
-		QueueChannel adapterErrors = context.getBean("adapterErrors", QueueChannel.class);
-
-		Message<RedisList<Object>> message = (Message<RedisList<Object>>) redisChannel.receive(1000);
-		assertNotNull(message);
-		assertNotNull(adapterErrors.receive(2000));
-		spca.stop();
-		context.close();
-	}
-
-
-	@Test
-	@RedisAvailable
-	@SuppressWarnings("unchecked")
-	public void testZsetInboundConfiguration(){
-		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.prepareZset(jcf);
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("zset-inbound-adapter.xml", this.getClass());
-		SourcePollingChannelAdapter zsetAdapterNoScore =
-				context.getBean("zsetAdapterNoScore", SourcePollingChannelAdapter.class);
-		zsetAdapterNoScore.start();
-
-		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
-
-		Message<RedisZSet<Object>> message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
-		assertNotNull(message);
-		assertEquals(13, message.getPayload().size());
-
-		//poll again, should get the same stuff
-		message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
-		assertNotNull(message);
-		assertEquals(13, message.getPayload().size());
-
-		zsetAdapterNoScore.stop();
-		context.close();
-	}
-
-	@Test
-	@RedisAvailable
-	@SuppressWarnings("unchecked")
-	public void testZsetInboundConfigurationWithScoreRange(){
-		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.prepareZset(jcf);
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("zset-inbound-adapter.xml", this.getClass());
-		SourcePollingChannelAdapter zsetAdapterWithScoreRange =
-				context.getBean("zsetAdapterWithScoreRange", SourcePollingChannelAdapter.class);
-		zsetAdapterWithScoreRange.start();
-
-		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
-
-		Message<RedisZSet<Object>> message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
-		assertNotNull(message);
-		assertEquals(11, message.getPayload().rangeByScore(18, 20).size());
-
-		//poll again, should get the same stuff
-		message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
-		assertNotNull(message);
-		assertEquals(11, message.getPayload().rangeByScore(18, 20).size());
-
-		zsetAdapterWithScoreRange.stop();
-		context.close();
-	}
-
-	@Test
-	@RedisAvailable
-	@SuppressWarnings("unchecked")
-	public void testZsetInboundConfigurationWithSingleScore(){
-		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.prepareZset(jcf);
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("zset-inbound-adapter.xml", this.getClass());
-		SourcePollingChannelAdapter zsetAdapterWithSingleScore =
-				context.getBean("zsetAdapterWithSingleScore", SourcePollingChannelAdapter.class);
-		zsetAdapterWithSingleScore.start();
-
-		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
-
-		Message<RedisZSet<Object>> message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
-		assertNotNull(message);
-		assertEquals(2, message.getPayload().rangeByScore(18, 18).size());
-
-		//poll again, should get the same stuff
-		message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
-		assertNotNull(message);
-		assertEquals(2, message.getPayload().rangeByScore(18, 18).size());
-
-		zsetAdapterWithSingleScore.stop();
-		context.close();
-	}
-
-	@Test
-	@RedisAvailable
-	@SuppressWarnings("unchecked")
-	public void testZsetInboundConfigurationWithSingleScoreAndSynchronization() throws Exception{
-		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
-		this.prepareZset(jcf);
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("zset-inbound-adapter.xml", this.getClass());
-		SourcePollingChannelAdapter zsetAdapterWithSingleScoreAndSynchronization =
-				context.getBean("zsetAdapterWithSingleScoreAndSynchronization", SourcePollingChannelAdapter.class);
-
-		SourcePollingChannelAdapter zsetAdapterNoScore =
-				context.getBean("zsetAdapterNoScore", SourcePollingChannelAdapter.class);
-
-		QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
-		QueueChannel otherRedisChannel = context.getBean("otherRedisChannel", QueueChannel.class);
-		// get all 13 presidents
-		zsetAdapterNoScore.start();
-
-		Message<RedisZSet<Object>> message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
-		assertNotNull(message);
-		assertEquals(13, message.getPayload().size());
-
-		zsetAdapterNoScore.stop();
-		Thread.sleep(1000);
-		// get only presidents for 18th century
-		zsetAdapterWithSingleScoreAndSynchronization.start();
-
-		message = (Message<RedisZSet<Object>>) otherRedisChannel.receive(1000);
-		assertNotNull(message);
-		assertEquals(2, message.getPayload().rangeByScore(18, 18).size());
-		//message.getPayload().remo
-		zsetAdapterWithSingleScoreAndSynchronization.stop();
-		Thread.sleep(1000);
-
-		// ... however other elements are still available 13-2=11
-		zsetAdapterNoScore.start();
-		message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
-		assertNotNull(message);
-		assertEquals(11, message.getPayload().size());
-
-		zsetAdapterNoScore.stop();
-		context.close();
+	public void validateFullConfiguration(){
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("inbound-store-adapter-parser.xml", this.getClass());
+		RedisStoreMessageSource source =
+				TestUtils.getPropertyValue(context.getBean("adapterWithConnectionFactory"), "source", RedisStoreMessageSource.class);
+		assertEquals(context.getBean("keySerializer"), TestUtils.getPropertyValue(source, "keySerializer"));
+		assertEquals(context.getBean("valueSerializer"), TestUtils.getPropertyValue(source, "valueSerializer"));
+		assertEquals(context.getBean("hashKeySerializer"), TestUtils.getPropertyValue(source, "hashKeySerializer"));
+		assertEquals(context.getBean("hashValueSerializer"), TestUtils.getPropertyValue(source, "hashValueSerializer"));
+		assertEquals("'presidents'", ((SpelExpression)TestUtils.getPropertyValue(source, "keyExpression")).getExpressionString());
+		assertEquals("LIST", ((CollectionType)TestUtils.getPropertyValue(source, "collectionType")).toString());
 	}
 
 	@Test(expected=BeanDefinitionParsingException.class)
-	public void testTemplateAndCfMutualExclusivity(){
-		new ClassPathXmlApplicationContext("inbound-template-cf-fail.xml", this.getClass());
+	public void validateFailureIfTemplateAndSerializers(){
+		new ClassPathXmlApplicationContext("inbound-store-adapter-parser-fail.xml", this.getClass());
 	}
 }

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

@@ -18,6 +18,7 @@ import static org.junit.Assert.assertEquals;
 
 import org.junit.Test;
 
+import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.data.redis.support.collections.RedisCollectionFactoryBean.CollectionType;
 import org.springframework.expression.common.LiteralExpression;
@@ -34,11 +35,16 @@ public class RedisCollectionsOutboundChannelAdapterParserTests {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter-parser.xml", this.getClass());
 		RedisCollectionPopulatingMessageHandler handler =
 				TestUtils.getPropertyValue(context.getBean("storeAdapter.adapter"), "handler", RedisCollectionPopulatingMessageHandler.class);
-		assertEquals(TestUtils.getPropertyValue(handler, "keySerializer"), context.getBean("keySerializer"));
-		assertEquals(TestUtils.getPropertyValue(handler, "valueSerializer"), context.getBean("valueSerializer"));
-		assertEquals(TestUtils.getPropertyValue(handler, "hashKeySerializer"), context.getBean("hashKeySerializer"));
-		assertEquals(TestUtils.getPropertyValue(handler, "hashValueSerializer"), context.getBean("hashValueSerializer"));
+		assertEquals(context.getBean("keySerializer"), TestUtils.getPropertyValue(handler, "keySerializer"));
+		assertEquals(context.getBean("valueSerializer"), TestUtils.getPropertyValue(handler, "valueSerializer"));
+		assertEquals(context.getBean("hashKeySerializer"), TestUtils.getPropertyValue(handler, "hashKeySerializer"));
+		assertEquals(context.getBean("hashValueSerializer"), TestUtils.getPropertyValue(handler, "hashValueSerializer"));
 		assertEquals("pepboys", ((LiteralExpression)TestUtils.getPropertyValue(handler, "keyExpression")).getExpressionString());
 		assertEquals("PROPERTIES", ((CollectionType)TestUtils.getPropertyValue(handler, "collectionType")).toString());
 	}
+
+	@Test(expected=BeanDefinitionParsingException.class)
+	public void validateFailureIfTemplateAndSerializers(){
+		new ClassPathXmlApplicationContext("store-outbound-adapter-parser-fail.xml", this.getClass());
+	}
 }

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

@@ -8,7 +8,8 @@
 
 	<int-redis:inbound-channel-adapter
 		id="adapter" topics="foo, bar" channel="receiveChannel" error-channel="testErrorChannel"
-		message-converter="testConverter" />
+		message-converter="testConverter" 
+		serializer="serializer"/>
 
 	<int:channel id="receiveChannel">
 		<int:queue />
@@ -26,6 +27,8 @@
 	<int-redis:inbound-channel-adapter
 		id="autoChannel" topics="foo, bar" error-channel="testErrorChannel"
 		message-converter="testConverter" />
+		
+	<bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
 
 	<int:bridge input-channel="autoChannel" output-channel="nullChannel"/>
 

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

@@ -26,9 +26,13 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ApplicationContext;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.redis.inbound.RedisInboundChannelAdapter;
+import org.springframework.integration.redis.inbound.RedisStoreMessageSource;
 import org.springframework.integration.redis.rules.RedisAvailable;
 import org.springframework.integration.redis.rules.RedisAvailableTests;
 import org.springframework.integration.support.converter.SimpleMessageConverter;
@@ -65,6 +69,7 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests{
 		assertEquals(errorChannelBean, accessor.getPropertyValue("errorChannel"));
 		Object converterBean = context.getBean("testConverter");
 		assertEquals(converterBean, accessor.getPropertyValue("messageConverter"));
+		assertEquals(context.getBean("serializer"), accessor.getPropertyValue("serializer"));
 	}
 
 	@Test
@@ -87,6 +92,46 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests{
 		assertSame(autoChannel, TestUtils.getPropertyValue(autoChannelAdapter, "outputChannel"));
 	}
 
+	@Test(expected=IllegalStateException.class)
+	public void cantSetKeySerializer() {
+		RedisTemplate<String, String> redisTemplate = new RedisTemplate<String, String>();
+		redisTemplate.setConnectionFactory(new JedisConnectionFactory());
+		RedisStoreMessageSource source = new RedisStoreMessageSource(redisTemplate,
+				new LiteralExpression("foo"));
+		source.setKeySerializer(new StringRedisSerializer());
+		source.afterPropertiesSet();
+	}
+
+	@Test(expected=IllegalStateException.class)
+	public void cantSetValueSerializer() {
+		RedisTemplate<String, String> redisTemplate = new RedisTemplate<String, String>();
+		redisTemplate.setConnectionFactory(new JedisConnectionFactory());
+		RedisStoreMessageSource source = new RedisStoreMessageSource(redisTemplate,
+				new LiteralExpression("foo"));
+		source.setValueSerializer(new StringRedisSerializer());
+		source.afterPropertiesSet();
+	}
+
+	@Test(expected=IllegalStateException.class)
+	public void cantSetHashKeySerializer() {
+		RedisTemplate<String, String> redisTemplate = new RedisTemplate<String, String>();
+		redisTemplate.setConnectionFactory(new JedisConnectionFactory());
+		RedisStoreMessageSource source = new RedisStoreMessageSource(redisTemplate,
+				new LiteralExpression("foo"));
+		source.setHashKeySerializer(new StringRedisSerializer());
+		source.afterPropertiesSet();
+	}
+
+	@Test(expected=IllegalStateException.class)
+	public void cantSetHashValueSerializer() {
+		RedisTemplate<String, String> redisTemplate = new RedisTemplate<String, String>();
+		redisTemplate.setConnectionFactory(new JedisConnectionFactory());
+		RedisStoreMessageSource source = new RedisStoreMessageSource(redisTemplate,
+				new LiteralExpression("foo"));
+		source.setHashValueSerializer(new StringRedisSerializer());
+		source.afterPropertiesSet();
+	}
+
 	@SuppressWarnings("unused")
 	private static class TestMessageConverter extends SimpleMessageConverter {
 	}

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

@@ -12,7 +12,8 @@
 	<int-redis:outbound-channel-adapter id="outboundAdapter"
 			channel="sendChannel"
 			topic="foo"
-			message-converter="testConverter"/>
+			message-converter="testConverter"
+			serializer="serializer"/>
 
 	<int-redis:inbound-channel-adapter channel="receiveChannel" topics="foo"/>
 
@@ -29,5 +30,7 @@
 	<int:chain input-channel="redisOutboudChain">
 		<int-redis:outbound-channel-adapter topic="foo"/>
 	</int:chain>
+	
+	<bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
 
 </beans>

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

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.redis.config;
 
+import static junit.framework.Assert.assertEquals;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -33,8 +35,6 @@ import org.springframework.integration.support.converter.SimpleMessageConverter;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
-import static junit.framework.Assert.assertEquals;
-
 /**
  * @author Oleg Zhurakousky
  * @author Mark Fisher
@@ -46,8 +46,8 @@ public class RedisOutboundChannelAdapterParserTests extends RedisAvailableTests{
 
 	@Autowired
 	private ApplicationContext context;
-	
-	@Test 
+
+	@Test
 	@RedisAvailable
 	public void validateConfiguration() {
 		EventDrivenConsumer adapter = context.getBean("outboundAdapter", EventDrivenConsumer.class);
@@ -58,9 +58,10 @@ public class RedisOutboundChannelAdapterParserTests extends RedisAvailableTests{
 		assertEquals("foo", accessor.getPropertyValue("defaultTopic"));
 		Object converterBean = context.getBean("testConverter");
 		assertEquals(converterBean, accessor.getPropertyValue("messageConverter"));
+		assertEquals(context.getBean("serializer"), accessor.getPropertyValue("serializer"));
 	}
- 
-	@Test 
+
+	@Test
 	@RedisAvailable
 	public void testOutboundChannelAdapterMessaging() throws Exception{
 		MessageChannel sendChannel = context.getBean("sendChannel", MessageChannel.class);

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-store-adapter-parser-fail.xml

@@ -0,0 +1,33 @@
+<?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"
+	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-2.2.xsd">
+
+	<int-redis:store-inbound-channel-adapter id="adapterWithTemplate"
+											redis-template="redisTemplate"
+											key-expression="'presidents'"
+											channel="redisChannel"
+											auto-startup="false"
+											collection-type="LIST"
+											key-serializer="serializer"
+											value-serializer="serializer"
+											hash-key-serializer="serializer"
+											hash-value-serializer="serializer">
+		<int:poller fixed-rate="2000" max-messages-per-poll="10"/>
+	</int-redis:store-inbound-channel-adapter>
+	
+	<bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate"/>
+
+	<int:channel id="redisChannel"/>
+	
+	<bean id="serializer" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer"/>
+	
+	<bean id="redisConnectionFactory" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.data.redis.connection.RedisConnectionFactory"/>
+	</bean>
+
+</beans>

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-store-adapter-parser.xml

@@ -0,0 +1,38 @@
+<?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"
+	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-2.2.xsd">
+
+	<int-redis:store-inbound-channel-adapter id="adapterWithConnectionFactory"
+											connection-factory="redisConnectionFactory"
+											key-expression="'presidents'"
+											channel="redisChannel"
+											auto-startup="false"
+											collection-type="LIST"
+											key-serializer="keySerializer"
+											value-serializer="valueSerializer"
+											hash-key-serializer="hashKeySerializer"
+											hash-value-serializer="hashValueSerializer">
+		<int:poller fixed-rate="2000" max-messages-per-poll="10"/>
+	</int-redis:store-inbound-channel-adapter>
+
+	<bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate">
+		<property name="connectionFactory" ref="redisConnectionFactory"/>
+	</bean>
+
+	<int:channel id="redisChannel"/>
+		
+	<bean id="keySerializer" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer"/>
+	<bean id="valueSerializer" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer"/>
+	<bean id="hashKeySerializer" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer"/>
+	<bean id="hashValueSerializer" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer"/>
+	
+	<bean id="redisConnectionFactory" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.data.redis.connection.RedisConnectionFactory"/>
+	</bean>
+
+</beans>

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/list-inbound-adapter.xml

@@ -72,12 +72,6 @@
 	</int:transaction-synchronization-factory>
 
 	<bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate">
-		<property name="keySerializer">
-			<bean class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
-		</property>
-		<property name="valueSerializer">
-			<bean class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer"/>
-		</property>
 		<property name="connectionFactory" ref="redisConnectionFactory"/>
 	</bean>
 

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/store-outbound-adapter-parser-fail.xml

@@ -0,0 +1,37 @@
+<?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:context="http://www.springframework.org/schema/context"
+	xmlns:tx="http://www.springframework.org/schema/tx"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		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
+		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
+
+	<int-redis:store-outbound-channel-adapter id="storeAdapter"
+											  redis-template="redisTemplate"
+											  collection-type="PROPERTIES"
+											  key="pepboys"
+											  key-serializer="keySerializer"
+											  value-serializer="valueSerializer"
+											  hash-key-serializer="hashKeySerializer"
+											  hash-value-serializer="hashValueSerializer"
+											  auto-startup="false"/>
+
+	<bean id="keySerializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
+	<bean id="valueSerializer" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer"/>
+	<bean id="hashKeySerializer" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer"/>
+	<bean id="hashValueSerializer" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer"/>
+
+	<bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate">
+		<property name="connectionFactory" ref="redisConnectionFactory"/>
+	</bean>
+	
+	<bean id="redisConnectionFactory" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.data.redis.connection.RedisConnectionFactory"/>
+	</bean>
+
+</beans>

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/store-outbound-adapter-parser.xml

@@ -25,8 +25,8 @@
 	<bean id="hashKeySerializer" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer"/>
 	<bean id="hashValueSerializer" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer"/>
 
-	<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
-		<property name="port" value="7379"/>
+	<bean id="redisConnectionFactory" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.data.redis.connection.RedisConnectionFactory"/>
 	</bean>
 
 </beans>

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/zset-inbound-adapter.xml

@@ -46,7 +46,7 @@
 	</int-redis:store-inbound-channel-adapter>
 	
 	<int:transaction-synchronization-factory id="syncFactory">
-		 <int:after-commit expression="#resource.attributes['store'].removeByScore(18, 18)"/>
+		 <int:after-commit expression="#store.removeByScore(18, 18)"/>
 	</int:transaction-synchronization-factory>
 
 	<int:channel id="redisChannel">

spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java

@@ -26,8 +26,6 @@ import org.springframework.data.redis.core.BoundListOperations;
 import org.springframework.data.redis.core.BoundZSetOperations;
 import org.springframework.data.redis.core.RedisCallback;
 import org.springframework.data.redis.core.RedisTemplate;
-import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
-import org.springframework.data.redis.serializer.StringRedisSerializer;
 
 /**
  * @author Oleg Zhurakousky
@@ -59,9 +57,7 @@ public class RedisAvailableTests {
 
 		RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<Object, Object>();
 		redisTemplate.setConnectionFactory(jcf);
-		redisTemplate.setKeySerializer(new StringRedisSerializer());
-		redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
-
+		redisTemplate.afterPropertiesSet();
 		BoundListOperations<Object, Object> ops = redisTemplate.boundListOps("presidents");
 
 		ops.rightPush("John Adams");
@@ -85,8 +81,7 @@ public class RedisAvailableTests {
 
 		RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<Object, Object>();
 		redisTemplate.setConnectionFactory(jcf);
-		redisTemplate.setKeySerializer(new StringRedisSerializer());
-		redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
+		redisTemplate.afterPropertiesSet();
 
 		BoundZSetOperations<Object, Object> ops = redisTemplate.boundZSetOps("presidents");