INT-2733/4/6 Configure Redis Serializers

Add support for serializers to the RedisStore Inbound adapters.

Fix the serializer support for RedisStore outbound adapters
to ensure that serializers cannot be set if adapter is
bootstrapped with a RedisTemplate.

Add support for serializers to Redis Inbound/Outbound adapters.
NOTE: unlike the RedisStore adapters Redis in/out adapters only
use one serializer, so the name of the attribute
is 'serializer' and, on the Redis Outbound adapter,
it corresponds to the RedisTemplate.valueSerializer.

INT-2733/4/6

Fix RedisCollectionPopulatingMessageHandler to
ensure that the checks for setting serializers also exist
at the class level in case an instance is created outside
of the namespace parser's control.

INT-2733/4/6 polishing

INT-2733/4/6 polishing

INT-2733/4/6 polishing

INT-2733/4/6 Polishing

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");