Merge pull request #577 from olegz/INT-2637c

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

@@ -38,7 +38,6 @@ public class RedisCollectionsInboundChannelAdapterParser extends AbstractPolling
 	@Override
 	protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(RedisStoreMessageSource.class);
-
 		String redisTemplate = element.getAttribute("redis-template");
 		String connectionFactory = element.getAttribute("connection-factory");
 		if (StringUtils.hasText(redisTemplate) && StringUtils.hasText(connectionFactory)){

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

@@ -0,0 +1,77 @@
+/*
+ * 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.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.ExpressionFactoryBean;
+import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.redis.outbound.RedisCollectionPopulatingMessageHandler;
+import org.springframework.util.StringUtils;
+import org.w3c.dom.Element;
+/**
+ * Parser for redis:store-outbound-channel-adapter element
+ *
+ * @author Oleg Zhurakousky
+ * @since 2.2
+ */
+public class RedisCollectionsOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
+
+	@Override
+	protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisCollectionPopulatingMessageHandler.class);
+
+		String redisTemplateRef = element.getAttribute("redis-template");
+		String connectionFactory = element.getAttribute("connection-factory");
+		if (StringUtils.hasText(redisTemplateRef) && StringUtils.hasText(connectionFactory)){
+			parserContext.getReaderContext().error("Only one of 'redis-template' or 'connection-factory'" +
+					" is allowed", element);
+		}
+
+		if (StringUtils.hasText(redisTemplateRef)){
+			builder.addConstructorArgReference(redisTemplateRef);
+		}
+		else {
+			if (!StringUtils.hasText(connectionFactory)) {
+				connectionFactory = "redisConnectionFactory";
+			}
+			builder.addConstructorArgReference(connectionFactory);
+		}
+
+		boolean atLeastOneRequired = false;
+		RootBeanDefinition expressionDef =
+				IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("key", "key-expression",
+						parserContext, element, atLeastOneRequired);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "collection-type");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload-elements");
+
+		if (expressionDef != null){
+			builder.addConstructorArgValue(expressionDef);
+		}
+
+		String mapKeyExpression = element.getAttribute("map-key-expression");
+		if (StringUtils.hasText(mapKeyExpression)) {
+			RootBeanDefinition mapKeyExpressionDef = new RootBeanDefinition(ExpressionFactoryBean.class);
+			mapKeyExpressionDef.getConstructorArgumentValues().addGenericArgumentValue(mapKeyExpression);
+			builder.addPropertyValue("mapKeyExpression", mapKeyExpressionDef);
+		}
+
+		return builder.getBeanDefinition();
+	}
+}

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

@@ -29,6 +29,7 @@ public class RedisNamespaceHandler extends AbstractIntegrationNamespaceHandler {
 		registerBeanDefinitionParser("publish-subscribe-channel", new RedisChannelParser());
 		registerBeanDefinitionParser("inbound-channel-adapter", new RedisInboundChannelAdapterParser());
 		registerBeanDefinitionParser("store-inbound-channel-adapter", new RedisCollectionsInboundChannelAdapterParser());
+		registerBeanDefinitionParser("store-outbound-channel-adapter", new RedisCollectionsOutboundChannelAdapterParser());
 		registerBeanDefinitionParser("outbound-channel-adapter", new RedisOutboundChannelAdapterParser());
 	}
 }

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

@@ -0,0 +1,452 @@
+/*
+ * 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.outbound;
+
+import java.util.Collection;
+import java.util.Map;
+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;
+import org.springframework.data.redis.core.BoundZSetOperations;
+import org.springframework.data.redis.core.RedisConnectionUtils;
+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.RedisCollectionFactoryBean;
+import org.springframework.data.redis.support.collections.RedisCollectionFactoryBean.CollectionType;
+import org.springframework.data.redis.support.collections.RedisList;
+import org.springframework.data.redis.support.collections.RedisMap;
+import org.springframework.data.redis.support.collections.RedisProperties;
+import org.springframework.data.redis.support.collections.RedisSet;
+import org.springframework.data.redis.support.collections.RedisStore;
+import org.springframework.data.redis.support.collections.RedisZSet;
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageHandlingException;
+import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.handler.AbstractMessageHandler;
+import org.springframework.integration.redis.support.RedisHeaders;
+import org.springframework.integration.util.ExpressionUtils;
+import org.springframework.util.Assert;
+import org.springframework.util.NumberUtils;
+
+/**
+ * Implementation of {@link MessageHandler} which writes Message data into a Redis store
+ * identified by a key {@link String}.
+ * It supports the collection types identified by {@link CollectionType}.
+ *
+ * It also supports batch updates and single item entry.
+ *
+ * "Batch updates" means that the payload of the Message may be a Map or Collection.
+ * With such a payload, individual items from it are added to the corresponding Redis store.
+ * See {@link #handleMessage(Message)} for more details.
+ *
+ * You can also choose to persist such a payload as a single item if the {@link #extractPayloadElements}
+ * property is set to false (default is true).
+ *
+ * @author Oleg Zhurakousky
+ * @author Gary Russell
+ * @since 2.2
+ */
+public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHandler {
+
+	private final Log logger = LogFactory.getLog(this.getClass());
+
+	private volatile StandardEvaluationContext evaluationContext;
+
+	private volatile Expression keyExpression =
+			new SpelExpressionParser().parseExpression("headers." + RedisHeaders.KEY);
+
+	private volatile Expression mapKeyExpression =
+			new SpelExpressionParser().parseExpression("headers." + RedisHeaders.MAP_KEY);
+
+	private volatile boolean mapKeyExpressionExplicitlySet;
+
+	private final RedisTemplate<String, ?> redisTemplate;
+
+	private volatile CollectionType collectionType = CollectionType.LIST;
+
+	private volatile boolean extractPayloadElements = true;
+
+	/**
+	 * Will construct this instance using fully created and initialized instance of
+	 * provided {@link RedisTemplate}
+	 *
+	 * The default expression 'headers.{@link RedisHeaders#KEY}'
+	 * will be used.
+	 * @param redisTemplate
+	 */
+	public RedisCollectionPopulatingMessageHandler(RedisTemplate<String, ?> redisTemplate) {
+		this(redisTemplate, null);
+	}
+
+	/**
+	 * Will construct this instance using
+	 * provided {@link RedisTemplate} and {@link #keyExpression}. The RedisTemplate must
+	 * be fully initialized.
+	 * If {@link #keyExpression} is null, the default expression 'headers.{@link RedisHeaders#KEY}'
+	 * will be used.
+	 *
+	 * @param redisTemplate
+	 * @param keyExpression
+	 */
+	public RedisCollectionPopulatingMessageHandler(RedisTemplate<String, ?> redisTemplate, Expression keyExpression) {
+		Assert.notNull(redisTemplate, "'redisTemplate' must not be null");
+		this.redisTemplate = redisTemplate;
+		if (keyExpression != null) {
+			this.keyExpression = keyExpression;
+		}
+	}
+
+	/**
+	 * Will construct this instance using the provided {@link RedisConnectionFactory}.
+	 * It will create an instance of {@link RedisTemplate}, initializing it with a
+	 * {@link StringRedisSerializer} for the keySerializer and a {@link JdkSeriaalizationRedisSerializer}
+	 * for each of valueSerializer, hasKeySerializer, and hashValueSerializer.
+	 *
+	 * The default expression 'headers.{@link RedisHeaders#KEY}'
+	 * will be used.
+	 * @param redisTemplate
+	 */
+	public RedisCollectionPopulatingMessageHandler(RedisConnectionFactory connectionFactory) {
+		this(connectionFactory, null);
+	}
+
+	/**
+	 * Will construct this instance using the provided {@link RedisConnectionFactory} and {@link #keyExpression}
+	 * It will create an instance of {@link RedisTemplate} initializing it with a
+	 * {@link StringRedisSerializer} for the keySerializer and a {@link JdkSerializationRedisSerializer}
+	 * for each of valueSerializer, hasKeySerializer, and hashValueSerializer.
+	 *
+	 * If {@link #keyExpression} is null, the default expression 'headers.{@link RedisHeaders#KEY}'
+	 * will be used.
+	 *
+	 * @param redisTemplate
+	 * @param keyExpression
+	 */
+	public RedisCollectionPopulatingMessageHandler(RedisConnectionFactory connectionFactory, Expression keyExpression) {
+		Assert.notNull(connectionFactory, "'connectionFactory' must not be null");
+
+		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;
+		if (keyExpression != null) {
+			this.keyExpression = keyExpression;
+		}
+	}
+
+	/**
+	 * Sets the collection type for this handler as per {@link CollectionType}
+	 *
+	 * @param collectionType
+	 */
+	public void setCollectionType(CollectionType collectionType) {
+		this.collectionType = collectionType;
+	}
+
+	/**
+	 * Sets the flag signifying that if the payload is a "multivalue" (i.e., Collection or Map),
+	 * it should be saved using addAll/putAll semantics. Default is 'true'.
+	 * If set to 'false' the payload will be saved as a single entry regardless of its type.
+	 * If the payload is not an instance of "multivalue" (i.e., Collection or Map)
+	 * the value of this attribute is meaningless as the payload will always be
+	 * stored as a single entry.
+	 *
+	 * @see #setExtractPayloadElements(boolean)
+	 *
+	 * @param extractPayloadElements
+	 */
+	public void setExtractPayloadElements(boolean extractPayloadElements) {
+		this.extractPayloadElements = extractPayloadElements;
+	}
+
+	/**
+	 * Sets the expression used as the key for Map and Properties entries.
+	 * Default is 'headers.{@link RedisHeaders#MAP_KEY}'
+	 * @param mapKeyExpression
+	 */
+	public void setMapKeyExpression(Expression mapKeyExpression) {
+		Assert.notNull(mapKeyExpression, "'mapKeyExpression' must not be null");
+		this.mapKeyExpression = mapKeyExpression;
+		this.mapKeyExpressionExplicitlySet = true;
+	}
+
+	@Override
+	public String getComponentType() {
+		return "redis:store-outbound-channel-adapter";
+	}
+
+	@Override
+	protected void onInit() throws Exception {
+		if (this.getBeanFactory() != null) {
+			this.evaluationContext =
+					ExpressionUtils.createStandardEvaluationContext(this.getBeanFactory());
+		}
+		else {
+			this.evaluationContext = ExpressionUtils.createStandardEvaluationContext();
+		}
+		Assert.state(!this.mapKeyExpressionExplicitlySet ||
+				(this.collectionType == CollectionType.MAP || this.collectionType == CollectionType.PROPERTIES),
+				"'mapKeyExpression' can only be set for CollectionType.MAP or CollectionType.PROPERTIES");
+	}
+
+	/**
+	 * Will extract payload from the Message storing it in the collection identified by the
+	 * {@link #collectionType}. The default CollectinType is LIST.
+	 * <p/>
+	 * The rules for storing payload are:
+	 * <p/>
+	 * <b>LIST/SET</b>
+	 * If payload is of type Collection and {@link #extractPayloadElements} is 'true' (default),
+	 * the payload will be added using the addAll() method. If {@link #extractPayloadElements} is set to 'false' then,
+	 * regardless of the payload type, the payload will be added using add();
+	 * <p/>
+	 * <b>ZSET</b>
+	 * In addition to rules described for LIST/SET, ZSET allows 'score' information
+	 * to be provided. The score can be provided using the {@link RedisHeaders#ZSET_SCORE} message header,
+	 * when the payload is a Collection, or
+	 * by sending a Map as the payload, where the Map 'key' is the value to be saved and the 'value' is
+	 * the score assigned to this value.
+	 * If {@link #extractPayloadElements} is set to 'false' the map will be stored as a single entry.
+	 * If the 'score' can not be determined, the default value (1) will be used.
+	 * <p/>
+	 * <b>MAP/PROPERTIES</b>
+	 * You can also store a payload of type Map or Properties following the same rules as above.
+	 * If payload itself needs to be stored as a value of the map/property then the map key must be
+	 * specified via the mapKeyExpression (default {@link RedisHeaders#MAP_KEY} Message header).
+	 */
+	@SuppressWarnings("unchecked")
+	@Override
+	protected void handleMessageInternal(Message<?> message) throws Exception {
+		String key = this.keyExpression.getValue(this.evaluationContext, message, String.class);
+
+		Assert.hasText(key, "Can not determine a 'key' for a Redis store. The key can be provided via the " +
+				"'key' or 'key-expression' atributes.");
+
+		RedisStore store = this.createStoreView(key);
+
+		try {
+			if (collectionType == CollectionType.ZSET) {
+				this.handleZset((RedisZSet<Object>) store, message);
+			}
+			else if (collectionType == CollectionType.SET) {
+				this.handleSet((RedisSet<Object>) store, message);
+			}
+			else if (collectionType == CollectionType.LIST) {
+				this.handleList((RedisList<Object>) store, message);
+			}
+			else if (collectionType == CollectionType.MAP) {
+				this.handleMap((RedisMap<Object, Object>) store, message);
+			}
+			else if (collectionType == CollectionType.PROPERTIES) {
+				this.handleProperties((RedisProperties) store, message);
+			}
+		}
+		catch (Exception e) {
+			throw new MessageHandlingException(message, "Failed to store Message data in Redis collection", e);
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private void handleZset(RedisZSet<Object> zset, final Message<?> message) throws Exception{
+		final Object payload = message.getPayload();
+
+		if (this.extractPayloadElements) {
+			final BoundZSetOperations<String, Object> ops =
+					(BoundZSetOperations<String, Object>) this.redisTemplate.boundZSetOps(zset.getKey());
+
+			if ((payload instanceof Map<?, ?> && this.isMapValuesOfTypeNumber((Map<?, ?>) payload))) {
+				final Map<Object, Number> pyloadAsMap = (Map<Object, Number>) payload;
+				this.processInPipeline(new PipelineCallback() {
+					public void process() {
+						for (Object key : pyloadAsMap.keySet()) {
+							Number d = pyloadAsMap.get(key);
+							ops.add(key, d == null ?
+									determineScore(message) :
+									NumberUtils.convertNumberToTargetClass(d, Double.class));
+						}
+					}
+				});
+			}
+			else if (payload instanceof Collection<?>) {
+				this.processInPipeline(new PipelineCallback() {
+					public void process() {
+						for (Object object : ((Collection<?>)payload)) {
+							ops.add(object, determineScore(message));
+						}
+					}
+				});
+			}
+			else {
+				this.addToZset(zset, payload, this.determineScore(message));
+			}
+		}
+		else {
+			this.addToZset(zset, payload, this.determineScore(message));
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private void handleList(RedisList<Object> list, Message<?> message) {
+		Object payload = message.getPayload();
+		if (this.extractPayloadElements) {
+			if (payload instanceof Collection<?>) {
+				list.addAll((Collection<? extends Object>) payload);
+			}
+			else {
+				list.add(payload);
+			}
+		}
+		else {
+			list.add(payload);
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private void handleSet(final RedisSet<Object> set, Message<?> message) {
+		final Object payload = message.getPayload();
+		if (this.extractPayloadElements && payload instanceof Collection<?>) {
+			final BoundSetOperations<String, Object> ops =
+					(BoundSetOperations<String, Object>) this.redisTemplate.boundSetOps(set.getKey());
+
+			this.processInPipeline(new PipelineCallback() {
+				public void process() {
+					for (Object object : ((Collection<?>)payload)) {
+						ops.add(object);
+					}
+				}
+			});
+		}
+		else {
+			set.add(payload);
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private void handleMap(final RedisMap<Object, Object> map, Message<?> message) {
+		final Object payload = message.getPayload();
+		if (this.extractPayloadElements && payload instanceof Map<?, ?>) {
+			this.processInPipeline(new PipelineCallback() {
+				public void process() {
+					map.putAll((Map<? extends Object, ? extends Object>) payload);
+				}
+			});
+		}
+		else {
+			Object key = this.assertMapEntry(message, false);
+			map.put(key, payload);
+		}
+	}
+
+	private void handleProperties(final RedisProperties properties, Message<?> message) {
+		final Object payload = message.getPayload();
+		if (this.extractPayloadElements && payload instanceof Properties) {
+			this.processInPipeline(new PipelineCallback() {
+				public void process() {
+					properties.putAll((Properties) payload);
+				}
+			});
+		}
+		else {
+			Object key = this.assertMapEntry(message, true);
+			properties.put(key, payload);
+		}
+	}
+
+	private void processInPipeline(PipelineCallback callback) {
+		RedisConnection connection =
+				RedisConnectionUtils.bindConnection(redisTemplate.getConnectionFactory());
+		try {
+			connection.openPipeline();
+			callback.process();
+		}
+		finally {
+			connection.closePipeline();
+			RedisConnectionUtils.unbindConnection(redisTemplate.getConnectionFactory());
+		}
+	}
+
+	private Object assertMapEntry(Message<?> message, boolean property) {
+		Object mapKey = this.mapKeyExpression.getValue(this.evaluationContext, message);
+		Assert.notNull(mapKey, "Can not determine a map key for the entry. The key is determined by evaluating " +
+				"the 'mapKeyExpression' property.");
+		Object payload = message.getPayload();
+		if (property) {
+			Assert.isInstanceOf(String.class, mapKey, "For property, key must be a String");
+			Assert.isInstanceOf(String.class, payload, "For property, payload must be a String");
+		}
+		Assert.isTrue(mapKey != null, "Failed to determine the key for the " +
+				"Redis Map entry. Payload is not a Map and '" + RedisHeaders.MAP_KEY +
+				"' header is not provided");
+		return mapKey;
+	}
+
+	private void addToZset(RedisZSet<Object> zset, Object objectToAdd, Double score) {
+		if (score != null) {
+			zset.add(objectToAdd, score);
+		}
+		else {
+			logger.debug("Zset Score could not be determined. Using default score of 1");
+			zset.add(objectToAdd);
+		}
+	}
+
+	private boolean isMapValuesOfTypeNumber(Map<?,?> map) {
+		for (Object value : map.values()) {
+			if (!(value instanceof Number)) {
+				logger.warn("Failed to extract payload elements because one of its values '" + value + "' is not of type Number");
+				return false;
+			}
+		}
+		return true;
+	}
+
+	private RedisStore createStoreView(String key) {
+		RedisCollectionFactoryBean fb = new RedisCollectionFactoryBean();
+		fb.setKey(key);
+		fb.setTemplate(this.redisTemplate);
+		fb.setType(this.collectionType);
+		fb.afterPropertiesSet();
+		return fb.getObject();
+	}
+
+	private double determineScore(Message<?> message) {
+		Object scoreHeader = message.getHeaders().get(RedisHeaders.ZSET_SCORE);
+		if (scoreHeader == null) {
+			return Double.valueOf(1);
+		}
+		else {
+			Assert.isInstanceOf(Number.class, scoreHeader, "Header " + RedisHeaders.ZSET_SCORE + " must be a Number");
+			Number score = (Number) scoreHeader;
+			return Double.valueOf(score.toString());
+		}
+	}
+
+	private interface PipelineCallback {
+		public void process();
+	}
+}

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

@@ -0,0 +1,20 @@
+package org.springframework.integration.redis.support;
+
+/**
+ * Pre-defined names and prefixes to be used for
+ * for dealing with headers required by Redis components
+ *
+ * @author Oleg Zhurakousky
+ * @since 2.2
+ */
+public class RedisHeaders {
+
+	public static final String PREFIX = "redis_";
+
+	public static final String KEY = PREFIX + "key";
+
+	public static final String MAP_KEY = PREFIX + "map_key";
+
+	public static final String ZSET_SCORE = PREFIX + "zset_score";
+
+}

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

@@ -0,0 +1,4 @@
+/**
+ * Provides supporting classes for this module.
+ */
+package org.springframework.integration.redis.support;

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

@@ -306,6 +306,29 @@
 		</xsd:complexType>
 	</xsd:element>
 
+	<xsd:element name="store-outbound-channel-adapter">
+		<xsd:annotation>
+			<xsd:documentation>
+	Defines Redis outbound channel adapter that writes the contents of the Message into
+	org.springframework.data.redis.support.collections.RedisStore
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="redisAdapterType">
+					<xsd:attributeGroup ref="storeAdapterAttributeGroup"/>
+					<xsd:attribute name="redis-template" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+							RedisTemplate to be used with this adapter
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
 	<xsd:complexType name="redisAdapterType">
 		<xsd:annotation>
 			<xsd:documentation>
@@ -346,7 +369,7 @@
 					<xsd:attribute name="channel" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation>
-			Channel to which Messages will be sent.
+			Channel to which Messages will be sent. 
 							</xsd:documentation>
 							<xsd:appinfo>
 								<tool:annotation kind="ref">
@@ -358,8 +381,8 @@
 					<xsd:attribute name="key-expression" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation>
-			SpEL expression which returns the name of the key for the collection being used. If you want to provide a
-			literal string value use the 'key' attribute.
+			SpEL expression that returns the name of the key for the collection being used. If you want to provide a
+			constant key, use the 'key' attribute.
 			This attribute is mutually exclusive with the 'key' attribute.
 							</xsd:documentation>
 						</xsd:annotation>
@@ -368,7 +391,7 @@
 						<xsd:annotation>
 							<xsd:documentation>
 			The name of the key for the collection being used. If you require a key to be dynamically
-			determined per each poll use the 'key-expression' attribute.
+			determined per each poll use 'key-expression' attribute. 
 			This attribute is mutually exclusive with the 'key-expression' attribute.
 							</xsd:documentation>
 						</xsd:annotation>
@@ -389,4 +412,90 @@
 			</xsd:complexContent>
 	</xsd:complexType>
 
+	<xsd:attributeGroup name="storeAdapterAttributeGroup">
+		<xsd:attribute name="key" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					Provide the name of the key for the collection being used. If you require a key that
+					is determined dynamically for each message, use the 'key-expression' attribute.
+					The default key is dynamically determined from the 'redis_key' header.
+					This attribute is mutually exclusive with the 'key-expression' attribute.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="key-expression" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					SpEL expression that returns the name of the key for the collection being used. If you want to provide a
+					constant key, use the 'key' attribute. Default is the 'redis_key' message header.
+					This attribute is mutually exclusive with the 'key' attribute.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="map-key-expression" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					SpEL expression that returns the name of the key for entry being stored. Only applies
+					if the 'collection-type' is MAP or PROPERTIES and 'extract-payload-elements' is false.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="extract-payload-elements" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+				If set to 'true' (Default) and the payload is an instance of a "multi-value" object (i.e., Collection or Map)
+				it will be stored using addAll/putAll semantics. Otherwise, if set to 'false' the payload will be stored
+				as single entry regardless of its type.
+				If the payload is not an instance of a "multi-value" object, the value of this attribute is ignored and
+				the payload will always be stored as a single entry.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="collection-type">
+			<xsd:annotation>
+				<xsd:documentation>
+					Collection type supported by this adapter
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:restriction base="xsd:string">
+					<xsd:enumeration value="LIST">
+						<xsd:annotation>
+							<xsd:documentation>
+								[DEFAULT] Redis List
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:enumeration>
+					<xsd:enumeration value="SET">
+						<xsd:annotation>
+							<xsd:documentation>
+								Redis Set
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:enumeration>
+					<xsd:enumeration value="ZSET">
+						<xsd:annotation>
+							<xsd:documentation>
+								Redis Sorted Set
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:enumeration>
+					<xsd:enumeration value="MAP">
+						<xsd:annotation>
+							<xsd:documentation>
+								Redis Map
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:enumeration>
+					<xsd:enumeration value="PROPERTIES">
+						<xsd:annotation>
+							<xsd:documentation>
+								Redis Properties
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:enumeration>
+				</xsd:restriction>
+			</xsd:simpleType>
+		</xsd:attribute>
+	</xsd:attributeGroup>
 </xsd:schema>

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

@@ -0,0 +1,342 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+import org.junit.Test;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.core.RedisOperations;
+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.DefaultRedisList;
+import org.springframework.data.redis.support.collections.DefaultRedisMap;
+import org.springframework.data.redis.support.collections.DefaultRedisSet;
+import org.springframework.data.redis.support.collections.DefaultRedisZSet;
+import org.springframework.data.redis.support.collections.RedisList;
+import org.springframework.data.redis.support.collections.RedisMap;
+import org.springframework.data.redis.support.collections.RedisProperties;
+import org.springframework.data.redis.support.collections.RedisSet;
+import org.springframework.data.redis.support.collections.RedisZSet;
+import org.springframework.expression.common.LiteralExpression;
+import org.springframework.expression.spel.standard.SpelExpression;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.MessageHandlingException;
+import org.springframework.integration.redis.outbound.RedisCollectionPopulatingMessageHandler;
+import org.springframework.integration.redis.rules.RedisAvailable;
+import org.springframework.integration.redis.rules.RedisAvailableTests;
+import org.springframework.integration.redis.support.RedisHeaders;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
+
+/**
+ * @author Oleg Zhurakousky
+ * @since 2.2
+ */
+public class RedisCollectionsOutboundChannelAdapterParserTests extends RedisAvailableTests {
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testListWithKeyAsHeader(){
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisList<String> redisList =
+				new DefaultRedisList<String>("pepboys",
+						(RedisOperations<String, String>) this.initTemplate(jcf, new RedisTemplate<String, String>()));
+		assertEquals(0, redisList.size());
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
+		MessageChannel redisChannel = context.getBean("listWithKeyAsHeader", MessageChannel.class);
+		List<String> pepboys = new ArrayList<String>();
+		pepboys.add("Manny");
+		pepboys.add("Moe");
+		pepboys.add("Jack");
+		Message<List<String>> message = MessageBuilder.withPayload(pepboys).setHeader("redis_key", "pepboys").build();
+		redisChannel.send(message);
+
+		assertEquals(3, redisList.size());
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testListWithProvidedKey(){
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisList<String> redisList =
+				new DefaultRedisList<String>("pepboys",
+						(RedisOperations<String, String>) this.initTemplate(jcf, new RedisTemplate<String, String>()));
+		assertEquals(0, redisList.size());
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
+		MessageChannel redisChannel = context.getBean("listWithKeyProvided", MessageChannel.class);
+		List<String> pepboys = new ArrayList<String>();
+		pepboys.add("Manny");
+		pepboys.add("Moe");
+		pepboys.add("Jack");
+		Message<List<String>> message = MessageBuilder.withPayload(pepboys).build();
+		redisChannel.send(message);
+
+		assertEquals(3, redisList.size());
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testMapToZsetWithProvidedKey(){
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisZSet<String> redisZset =
+				new DefaultRedisZSet<String>("presidents",
+						(RedisOperations<String, String>) this.initTemplate(jcf, new RedisTemplate<String, String>()));
+		assertEquals(0, redisZset.size());
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
+		MessageChannel redisChannel = context.getBean("mapToZset", MessageChannel.class);
+		Map<String, Integer> presidents = new HashMap<String, Integer>();
+		presidents.put("John Adams", 18);
+
+		presidents.put("Barack Obama", 21);
+		presidents.put("Thomas Jefferson", 19);
+		presidents.put("John Quincy Adams", 19);
+		presidents.put("Zachary Taylor", 19);
+
+		Message<Map<String, Integer>> message = MessageBuilder.withPayload(presidents).build();
+		redisChannel.send(message);
+
+		assertEquals(5, redisZset.size());
+		assertEquals(1, redisZset.rangeByScore(18, 18).size());
+		assertEquals(4, redisZset.rangeByScore(18, 19).size());
+
+		RedisCollectionPopulatingMessageHandler handler = context.getBean("mapToZset.handler",
+				RedisCollectionPopulatingMessageHandler.class);
+		assertEquals("'presidents'", TestUtils.getPropertyValue(handler, "keyExpression", SpelExpression.class).getExpressionString());
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testMapToMapWithProvidedKey(){
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisMap<String, String> redisMap =
+				new DefaultRedisMap<String, String>("pepboys",
+						(RedisOperations<String, ?>) this.initTemplate(jcf, new RedisTemplate<String, Map<String, String>>()));
+
+		assertEquals(0, redisMap.size());
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
+		MessageChannel redisChannel = context.getBean("mapToMapA", MessageChannel.class);
+		Map<String, String> pepboys = new HashMap<String, String>();
+		pepboys.put("1", "Manny");
+		pepboys.put("2", "Moe");
+		pepboys.put("3", "Jack");
+
+
+		Message<Map<String, String>> message = MessageBuilder.withPayload(pepboys).build();
+		redisChannel.send(message);
+		assertEquals("Manny", redisMap.get("1"));
+		assertEquals("Moe", redisMap.get("2"));
+		assertEquals("Jack", redisMap.get("3"));
+
+		RedisCollectionPopulatingMessageHandler handler = context.getBean("mapToMapA.handler",
+				RedisCollectionPopulatingMessageHandler.class);
+		assertEquals("pepboys", TestUtils.getPropertyValue(handler, "keyExpression", LiteralExpression.class).getExpressionString());
+		assertEquals("'foo'", TestUtils.getPropertyValue(handler, "mapKeyExpression", SpelExpression.class).getExpressionString());
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test(expected=MessageHandlingException.class)// map key is not proivided
+	@RedisAvailable
+	public void testMapToMapAsSingleEntryWithKeyAsHeaderFail(){
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisMap<String, Map<String, String>> redisMap =
+				new DefaultRedisMap<String, Map<String, String>>("pepboys",
+						(RedisOperations<String, ?>) this.initTemplate(jcf, new RedisTemplate<String, Map<String, Map<String, String>>>()));
+
+		assertEquals(0, redisMap.size());
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
+		MessageChannel redisChannel = context.getBean("mapToMapB", MessageChannel.class);
+		Map<String, String> pepboys = new HashMap<String, String>();
+		pepboys.put("1", "Manny");
+		pepboys.put("2", "Moe");
+		pepboys.put("3", "Jack");
+
+
+		Message<Map<String, String>> message = MessageBuilder.withPayload(pepboys).
+				setHeader(RedisHeaders.KEY, "pepboys").build();
+		redisChannel.send(message);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test(expected=MessageHandlingException.class)//key is not provided
+	@RedisAvailable
+	public void testMapToMapNoKey(){
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisMap<String, Map<String, String>> redisMap =
+				new DefaultRedisMap<String, Map<String, String>>("pepboys",
+						(RedisOperations<String, ?>) this.initTemplate(jcf, new RedisTemplate<String, Map<String, Map<String, String>>>()));
+
+		assertEquals(0, redisMap.size());
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
+		MessageChannel redisChannel = context.getBean("mapToMapB", MessageChannel.class);
+		Map<String, String> pepboys = new HashMap<String, String>();
+		pepboys.put("1", "Manny");
+		pepboys.put("2", "Moe");
+		pepboys.put("3", "Jack");
+
+
+		Message<Map<String, String>> message = MessageBuilder.withPayload(pepboys).build();
+		redisChannel.send(message);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testMapToMapAsSingleEntryWithKeyAsHeader(){
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisMap<String, Map<String, String>> redisMap =
+				new DefaultRedisMap<String, Map<String, String>>("pepboys",
+						(RedisOperations<String, ?>) this.initTemplate(jcf, new RedisTemplate<String, Map<String, Map<String, String>>>()));
+
+		assertEquals(0, redisMap.size());
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
+		MessageChannel redisChannel = context.getBean("mapToMapB", MessageChannel.class);
+		Map<String, String> pepboys = new HashMap<String, String>();
+		pepboys.put("1", "Manny");
+		pepboys.put("2", "Moe");
+		pepboys.put("3", "Jack");
+
+
+		Message<Map<String, String>> message = MessageBuilder.withPayload(pepboys).
+				setHeader(RedisHeaders.KEY, "pepboys").setHeader(RedisHeaders.MAP_KEY, "foo").build();
+		redisChannel.send(message);
+		Map<String, String> pepboyz = redisMap.get("foo");
+
+		assertEquals("Manny", pepboyz.get("1"));
+		assertEquals("Moe", pepboyz.get("2"));
+		assertEquals("Jack", pepboyz.get("3"));
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testSetWithKeyAsHeader(){
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisSet<String> redisList =
+				new DefaultRedisSet<String>("pepboys",
+						(RedisOperations<String, String>) this.initTemplate(jcf, new RedisTemplate<String, String>()));
+		assertEquals(0, redisList.size());
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
+		MessageChannel redisChannel = context.getBean("set", MessageChannel.class);
+		Set<String> pepboys = new HashSet<String>();
+		pepboys.add("Manny");
+		pepboys.add("Moe");
+		pepboys.add("Jack");
+		Message<Set<String>> message = MessageBuilder.withPayload(pepboys).setHeader("redis_key", "pepboys").build();
+		redisChannel.send(message);
+
+		assertEquals(3, redisList.size());
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testSetWithKeyAsHeaderNotParsed(){
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisSet<String> redisList =
+				new DefaultRedisSet<String>("pepboys",
+						(RedisOperations<String, String>) this.initTemplate(jcf, new RedisTemplate<String, String>()));
+		assertEquals(0, redisList.size());
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
+		MessageChannel redisChannel = context.getBean("setNotParsed", MessageChannel.class);
+		Set<String> pepboys = new HashSet<String>();
+		pepboys.add("Manny");
+		pepboys.add("Moe");
+		pepboys.add("Jack");
+		Message<Set<String>> message = MessageBuilder.withPayload(pepboys).setHeader("redis_key", "pepboys").build();
+		redisChannel.send(message);
+
+		assertEquals(1, redisList.size());
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testPojoIntoSet(){
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisSet<String> redisList =
+				new DefaultRedisSet<String>("pepboys",
+						(RedisOperations<String, String>) this.initTemplate(jcf, new RedisTemplate<String, String>()));
+		assertEquals(0, redisList.size());
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
+		MessageChannel redisChannel = context.getBean("pojoIntoSet", MessageChannel.class);
+		String pepboy = "Manny";
+		Message<String> message = MessageBuilder.withPayload(pepboy).setHeader("redis_key", "pepboys").build();
+		redisChannel.send(message);
+
+		assertEquals(1, redisList.size());
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testProperty(){
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisProperties redisProperties =
+				new RedisProperties("pepboys",
+						(RedisOperations<String, ?>) this.initTemplate(jcf, new RedisTemplate<String, Properties>()));
+
+		assertEquals(0, redisProperties.size());
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass());
+		MessageChannel redisChannel = context.getBean("property", MessageChannel.class);
+		Properties pepboys = new Properties();
+		pepboys.put("1", "Manny");
+		pepboys.put("2", "Moe");
+		pepboys.put("3", "Jack");
+
+		Message<Properties> message = MessageBuilder.withPayload(pepboys).build();
+		redisChannel.send(message);
+		assertEquals("Manny", redisProperties.get("1"));
+		assertEquals("Moe", redisProperties.get("2"));
+		assertEquals("Jack", redisProperties.get("3"));
+	}
+
+	private RedisTemplate<?,?> initTemplate(RedisConnectionFactory rcf, RedisTemplate<?, ?> redisTemplate){
+		redisTemplate.setConnectionFactory(rcf);
+		redisTemplate.setKeySerializer(new StringRedisSerializer());
+		redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
+		redisTemplate.setHashValueSerializer(new JdkSerializationRedisSerializer());
+		redisTemplate.setHashKeySerializer(new JdkSerializationRedisSerializer());
+		return redisTemplate;
+	}
+}

spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-template-cf-fail.xml

@@ -9,7 +9,7 @@
 
 
 	<int-redis:store-inbound-channel-adapter id="listAdapterWithSynchronizationAndRedisTemplate"
-											redis-template="redisTemplate"
+											redis-template="redisTemplate" 
 											connection-factory="redisConnectionFactory"
 											key-expression="'presidents'"
 											auto-startup="false">

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

@@ -0,0 +1,52 @@
+<?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-2.2.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-2.2.xsd
+		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.1.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+
+	<int-redis:store-outbound-channel-adapter id="listWithKeyAsHeader"/>
+
+	<int-redis:store-outbound-channel-adapter id="listWithKeyProvided" 
+											  collection-type="LIST"
+											  key="pepboys"/>
+
+	<int-redis:store-outbound-channel-adapter id="mapToZset"
+											  collection-type="ZSET"
+											  key-expression="'presidents'"/>
+
+	<int-redis:store-outbound-channel-adapter id="mapToMapA"
+											  collection-type="MAP"
+											  map-key-expression="'foo'"
+											  key="pepboys"/>
+
+	<int-redis:store-outbound-channel-adapter id="mapToMapB"
+											  collection-type="MAP"
+											  extract-payload-elements="false"/>
+
+	<int-redis:store-outbound-channel-adapter id="set"
+											  collection-type="SET"/>
+
+	<int-redis:store-outbound-channel-adapter id="setNotParsed"
+											  collection-type="SET"
+											  extract-payload-elements="false"/>
+
+	<int-redis:store-outbound-channel-adapter id="pojoIntoSet"
+											  collection-type="SET"/>
+
+	<int-redis:store-outbound-channel-adapter id="property"
+											  collection-type="PROPERTIES"
+											  key="pepboys"/>
+
+	<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
+		<property name="port" value="7379"/>
+	</bean>
+
+</beans>

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

@@ -45,7 +45,6 @@
 		</int:poller>
 	</int-redis:store-inbound-channel-adapter>
 
-
 	<int:channel id="redisChannel">
 		<int:queue/>
 	</int:channel>

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

@@ -0,0 +1,422 @@
+package org.springframework.integration.redis.outbound;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.junit.Test;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.core.RedisOperations;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
+import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.data.redis.support.collections.DefaultRedisList;
+import org.springframework.data.redis.support.collections.DefaultRedisZSet;
+import org.springframework.data.redis.support.collections.RedisCollectionFactoryBean.CollectionType;
+import org.springframework.data.redis.support.collections.RedisList;
+import org.springframework.data.redis.support.collections.RedisZSet;
+import org.springframework.expression.common.LiteralExpression;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageHandlingException;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.redis.rules.RedisAvailable;
+import org.springframework.integration.redis.rules.RedisAvailableTests;
+import org.springframework.integration.support.MessageBuilder;
+
+public class RedisCollectionPopulatingMessageHandlerTests extends RedisAvailableTests{
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testListWithListPayloadParsedAndProvidedKey() {
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		String key = "foo";
+		RedisList<String> redisList =
+				new DefaultRedisList<String>(key,
+						(RedisOperations<String, String>) this.initTemplate(jcf, new RedisTemplate<String, String>()));
+
+		assertEquals(0, redisList.size());
+
+		RedisCollectionPopulatingMessageHandler handler =
+				new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key));
+
+		List<String> list = new ArrayList<String>();
+		list.add("Manny");
+		list.add("Moe");
+		list.add("Jack");
+		Message<List<String>> message = new GenericMessage<List<String>>(list);
+		handler.handleMessage(message);
+
+		assertEquals(3, redisList.size());
+		assertEquals("Manny", redisList.get(0));
+		assertEquals("Moe", redisList.get(1));
+		assertEquals("Jack", redisList.get(2));
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testListWithListPayloadParsedAndProvidedKeyAsHeader() {
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		String key = "foo";
+		RedisList<String> redisList =
+				new DefaultRedisList<String>(key,
+						(RedisOperations<String, String>) this.initTemplate(jcf, new RedisTemplate<String, String>()));
+
+		assertEquals(0, redisList.size());
+
+		RedisCollectionPopulatingMessageHandler handler =
+				new RedisCollectionPopulatingMessageHandler(jcf, null);
+		handler.afterPropertiesSet();
+
+		List<String> list = new ArrayList<String>();
+		list.add("Manny");
+		list.add("Moe");
+		list.add("Jack");
+		Message<List<String>> message = MessageBuilder.withPayload(list).setHeader("redis_key", key).build();
+		handler.handleMessage(message);
+
+		assertEquals(3, redisList.size());
+		assertEquals("Manny", redisList.get(0));
+		assertEquals("Moe", redisList.get(1));
+		assertEquals("Jack", redisList.get(2));
+	}
+
+	@SuppressWarnings("unchecked")
+	@RedisAvailable
+	@Test(expected=MessageHandlingException.class)
+	public void testListWithListPayloadParsedAndNoKey() {
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		String key = "foo";
+		RedisList<String> redisList =
+				new DefaultRedisList<String>(key,
+						(RedisOperations<String, String>) this.initTemplate(jcf, new RedisTemplate<String, String>()));
+
+		assertEquals(0, redisList.size());
+
+		RedisCollectionPopulatingMessageHandler handler =
+				new RedisCollectionPopulatingMessageHandler(jcf, null);
+
+		List<String> list = new ArrayList<String>();
+		list.add("Manny");
+		list.add("Moe");
+		list.add("Jack");
+		Message<List<String>> message = MessageBuilder.withPayload(list).build();
+		handler.handleMessage(message);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testListWithListPayloadAsSingleEntry() {
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		String key = "foo";
+		RedisList<List<String>> redisList =
+				new DefaultRedisList<List<String>>(key,
+						(RedisOperations<String, List<String>>) this.initTemplate(jcf, new RedisTemplate<String, List<String>>()));
+
+		assertEquals(0, redisList.size());
+
+		RedisCollectionPopulatingMessageHandler handler =
+				new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key));
+		handler.setExtractPayloadElements(false);
+
+		List<String> list = new ArrayList<String>();
+		list.add("Manny");
+		list.add("Moe");
+		list.add("Jack");
+		Message<List<String>> message = new GenericMessage<List<String>>(list);
+		handler.handleMessage(message);
+
+		assertEquals(1, redisList.size());
+		List<String> resultList = redisList.get(0);
+		assertEquals("Manny", resultList.get(0));
+		assertEquals("Moe", resultList.get(1));
+		assertEquals("Jack", resultList.get(2));
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testZsetWithListPayloadParsedAndProvidedKeyDefaultScore() {
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		String key = "foo";
+		RedisZSet<String> redisZset =
+				new DefaultRedisZSet<String>(key,
+						(RedisOperations<String, String>) this.initTemplate(jcf, new RedisTemplate<String, String>()));
+
+		assertEquals(0, redisZset.size());
+
+		RedisCollectionPopulatingMessageHandler handler =
+				new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key));
+
+		handler.setCollectionType(CollectionType.ZSET);
+
+		List<String> list = new ArrayList<String>();
+		list.add("Manny");
+		list.add("Moe");
+		list.add("Jack");
+		Message<List<String>> message = new GenericMessage<List<String>>(list);
+		handler.handleMessage(message);
+
+		assertEquals(3, redisZset.size());
+		Set<TypedTuple<String>> pepboys = redisZset.rangeByScoreWithScores(1, 1);
+		for (TypedTuple<String> pepboy : pepboys) {
+			assertTrue(pepboy.getScore() == 1);
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testZsetWithListPayloadAsSingleEntryAndHeaderKeyHeaderScore() {
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		String key = "foo";
+		RedisZSet<List<String>> redisZset =
+				new DefaultRedisZSet<List<String>>(key,
+						(RedisOperations<String, List<String>>) this.initTemplate(jcf, new RedisTemplate<String, List<String>>()));
+
+		assertEquals(0, redisZset.size());
+
+		RedisCollectionPopulatingMessageHandler handler =
+				new RedisCollectionPopulatingMessageHandler(jcf, null);
+		handler.afterPropertiesSet();
+
+		handler.setCollectionType(CollectionType.ZSET);
+		handler.setExtractPayloadElements(false);
+
+		List<String> list = new ArrayList<String>();
+		list.add("Manny");
+		list.add("Moe");
+		list.add("Jack");
+		Message<List<String>> message = MessageBuilder.withPayload(list).setHeader("redis_key", key).
+						setHeader("redis_zset_score", 4).build();
+		handler.handleMessage(message);
+
+		assertEquals(1, redisZset.size());
+		Set<TypedTuple<List<String>>> entries = redisZset.rangeByScoreWithScores(1, 4);
+		for (TypedTuple<List<String>> pepboys : entries) {
+			assertTrue(pepboys.getScore() == 4);
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testZsetWithMapPayloadParsedHeaderKey() {
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		String key = "presidents";
+		RedisZSet<String> redisZset =
+				new DefaultRedisZSet<String>(key,
+						(RedisOperations<String, String>) this.initTemplate(jcf, new RedisTemplate<String, String>()));
+
+		assertEquals(0, redisZset.size());
+
+		RedisCollectionPopulatingMessageHandler handler =
+				new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key));
+
+		handler.setCollectionType(CollectionType.ZSET);
+
+		Map<String, Double> presidents = new HashMap<String, Double>();
+		presidents.put("John Adams", 18D);
+
+		presidents.put("Barack Obama", 21D);
+		presidents.put("Thomas Jefferson", 19D);
+		presidents.put("John Quincy Adams", 19D);
+		presidents.put("Zachary Taylor", 19D);
+
+		presidents.put("Theodore Roosevelt", 20D);
+		presidents.put("Woodrow Wilson", 20D);
+		presidents.put("George W. Bush", 21D);
+		presidents.put("Franklin D. Roosevelt", 20D);
+		presidents.put("Ronald Reagan", 20D);
+		presidents.put("William J. Clinton", 20D);
+		presidents.put("Abraham Lincoln", 19D);
+		presidents.put("George Washington", 18D);
+
+		Message<Map<String, Double>> message = MessageBuilder.withPayload(presidents).setHeader("redis_key", key).build();
+		handler.handleMessage(message);
+
+		assertEquals(13, redisZset.size());
+
+		Set<TypedTuple<String>> entries = redisZset.rangeByScoreWithScores(18, 19);
+		assertEquals(6, entries.size());
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testZsetWithMapPayloadPojoParsedHeaderKey() {
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		String key = "presidents";
+		RedisZSet<President> redisZset =
+				new DefaultRedisZSet<President>(key,
+						(RedisOperations<String, President>) this.initTemplate(jcf, new RedisTemplate<String, President>()));
+
+		assertEquals(0, redisZset.size());
+
+		RedisCollectionPopulatingMessageHandler handler =
+				new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key));
+
+		handler.setCollectionType(CollectionType.ZSET);
+
+		Map<President, Double> presidents = new HashMap<President, Double>();
+		presidents.put(new President("John Adams"), 18D);
+
+		presidents.put(new President("Barack Obama"), 21D);
+		presidents.put(new President("Thomas Jefferson"), 19D);
+		presidents.put(new President("John Quincy Adams"), 19D);
+		presidents.put(new President("Zachary Taylor"), 19D);
+
+		presidents.put(new President("Theodore Roosevelt"), 20D);
+		presidents.put(new President("Woodrow Wilson"), 20D);
+		presidents.put(new President("George W. Bush"), 21D);
+		presidents.put(new President("Franklin D. Roosevelt"), 20D);
+		presidents.put(new President("Ronald Reagan"), 20D);
+		presidents.put(new President("William J. Clinton"), 20D);
+		presidents.put(new President("Abraham Lincoln"), 19D);
+		presidents.put(new President("George Washington"), 18D);
+
+		Message<Map<President, Double>> message = MessageBuilder.withPayload(presidents).setHeader("redis_key", key).build();
+		handler.handleMessage(message);
+
+		assertEquals(13, redisZset.size());
+
+		Set<TypedTuple<President>> entries = redisZset.rangeByScoreWithScores(18, 19);
+		assertEquals(6, entries.size());
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testZsetWithMapPayloadPojoAsSingleEntryHeaderKey() {
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		String key = "presidents";
+		RedisZSet<Map<President, Double>> redisZset =
+				new DefaultRedisZSet<Map<President, Double>>(key,
+						(RedisOperations<String, Map<President, Double>>) this.initTemplate(jcf, new RedisTemplate<String, Map<President, Double>>()));
+
+		assertEquals(0, redisZset.size());
+
+		RedisCollectionPopulatingMessageHandler handler =
+				new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key));
+
+		handler.setCollectionType(CollectionType.ZSET);
+		handler.setExtractPayloadElements(false);
+
+		Map<President, Double> presidents = new HashMap<President, Double>();
+		presidents.put(new President("John Adams"), 18D);
+
+		presidents.put(new President("Barack Obama"), 21D);
+		presidents.put(new President("Thomas Jefferson"), 19D);
+
+		Message<Map<President, Double>> message = MessageBuilder.withPayload(presidents).setHeader("redis_key", key).build();
+		handler.handleMessage(message);
+
+		assertEquals(1, redisZset.size());
+	}
+
+	@Test(expected=IllegalStateException.class)
+	@RedisAvailable
+	public void testListWithMapKeyExpression() {
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		String key = "foo";
+		RedisCollectionPopulatingMessageHandler handler =
+				new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key));
+		handler.setMapKeyExpression(new LiteralExpression(key));
+		handler.afterPropertiesSet();
+	}
+
+	@Test(expected=IllegalStateException.class)
+	@RedisAvailable
+	public void testSetWithMapKeyExpression() {
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		String key = "foo";
+		RedisCollectionPopulatingMessageHandler handler =
+				new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key));
+		handler.setCollectionType(CollectionType.SET);
+		handler.setMapKeyExpression(new LiteralExpression(key));
+		handler.afterPropertiesSet();
+	}
+
+	@Test(expected=IllegalStateException.class)
+	@RedisAvailable
+	public void testZsetWithMapKeyExpression() {
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		String key = "foo";
+		RedisCollectionPopulatingMessageHandler handler =
+				new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key));
+		handler.setCollectionType(CollectionType.ZSET);
+		handler.setMapKeyExpression(new LiteralExpression(key));
+		handler.afterPropertiesSet();
+	}
+
+	@Test
+	@RedisAvailable
+	public void testMapWithMapKeyExpression() {
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		String key = "foo";
+		RedisCollectionPopulatingMessageHandler handler =
+				new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key));
+		handler.setCollectionType(CollectionType.MAP);
+		handler.setMapKeyExpression(new LiteralExpression(key));
+		try {
+			handler.afterPropertiesSet();
+		}
+		catch (Exception e) {
+			fail("No exception expected:" + e.getMessage());
+		}
+	}
+
+	@Test
+	@RedisAvailable
+	public void testPropertiesWithMapKeyExpression() {
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		String key = "foo";
+		RedisCollectionPopulatingMessageHandler handler =
+				new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key));
+		handler.setCollectionType(CollectionType.PROPERTIES);
+		handler.setMapKeyExpression(new LiteralExpression(key));
+		try {
+			handler.afterPropertiesSet();
+		}
+		catch (Exception e) {
+			fail("No exception expected:" + e.getMessage());
+		}
+	}
+
+	private RedisTemplate<?,?> initTemplate(RedisConnectionFactory rcf, RedisTemplate<?, ?> redisTemplate) {
+		redisTemplate.setConnectionFactory(rcf);
+		redisTemplate.setKeySerializer(new StringRedisSerializer());
+		redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
+		return redisTemplate;
+	}
+
+	private static class President implements Serializable {
+		private static final long serialVersionUID = 1L;
+		private String name;
+
+		public President(String name) {
+			this.name = name;
+		}
+
+		@SuppressWarnings("unused")
+		public String getName() {
+			return name;
+		}
+
+		@SuppressWarnings("unused")
+		public void setName(String name) {
+			this.name = name;
+		}
+	}
+}