diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParser.java index acd69becf7..aaa821a292 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParser.java +++ b/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)){ diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParser.java new file mode 100644 index 0000000000..929a9e9c25 --- /dev/null +++ b/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(); + } +} \ No newline at end of file diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java index 4c67c2e7e3..749ce9f1e1 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java +++ b/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()); } } \ No newline at end of file diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisCollectionPopulatingMessageHandler.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisCollectionPopulatingMessageHandler.java new file mode 100644 index 0000000000..c05bcbf814 --- /dev/null +++ b/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 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 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 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 redisTemplate = new RedisTemplate(); + 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. + *

+ * The rules for storing payload are: + *

+ * LIST/SET + * 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(); + *

+ * ZSET + * 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. + *

+ * MAP/PROPERTIES + * 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) store, message); + } + else if (collectionType == CollectionType.SET) { + this.handleSet((RedisSet) store, message); + } + else if (collectionType == CollectionType.LIST) { + this.handleList((RedisList) store, message); + } + else if (collectionType == CollectionType.MAP) { + this.handleMap((RedisMap) 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 zset, final Message message) throws Exception{ + final Object payload = message.getPayload(); + + if (this.extractPayloadElements) { + final BoundZSetOperations ops = + (BoundZSetOperations) this.redisTemplate.boundZSetOps(zset.getKey()); + + if ((payload instanceof Map && this.isMapValuesOfTypeNumber((Map) payload))) { + final Map pyloadAsMap = (Map) 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 list, Message message) { + Object payload = message.getPayload(); + if (this.extractPayloadElements) { + if (payload instanceof Collection) { + list.addAll((Collection) payload); + } + else { + list.add(payload); + } + } + else { + list.add(payload); + } + } + + @SuppressWarnings("unchecked") + private void handleSet(final RedisSet set, Message message) { + final Object payload = message.getPayload(); + if (this.extractPayloadElements && payload instanceof Collection) { + final BoundSetOperations ops = + (BoundSetOperations) 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 map, Message message) { + final Object payload = message.getPayload(); + if (this.extractPayloadElements && payload instanceof Map) { + this.processInPipeline(new PipelineCallback() { + public void process() { + map.putAll((Map) 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 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(); + } +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/RedisHeaders.java new file mode 100644 index 0000000000..7fbc664a51 --- /dev/null +++ b/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"; + +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/package-info.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/support/package-info.java new file mode 100644 index 0000000000..4ea4530ff2 --- /dev/null +++ b/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; \ No newline at end of file diff --git a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd index a045d2505c..8081ad9542 100644 --- a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd +++ b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd @@ -306,6 +306,29 @@ + + + + Defines Redis outbound channel adapter that writes the contents of the Message into + org.springframework.data.redis.support.collections.RedisStore + + + + + + + + + + RedisTemplate to be used with this adapter + + + + + + + + @@ -346,7 +369,7 @@ - Channel to which Messages will be sent. + Channel to which Messages will be sent. @@ -358,8 +381,8 @@ - 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. @@ -368,7 +391,7 @@ 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. @@ -389,4 +412,90 @@ + + + + + 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. + + + + + + + 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. + + + + + + + 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. + + + + + + + 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. + + + + + + + Collection type supported by this adapter + + + + + + + + [DEFAULT] Redis List + + + + + + + Redis Set + + + + + + + Redis Sorted Set + + + + + + + Redis Map + + + + + + + Redis Properties + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParserTests.java new file mode 100644 index 0000000000..91c92a8ec2 --- /dev/null +++ b/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 redisList = + new DefaultRedisList("pepboys", + (RedisOperations) this.initTemplate(jcf, new RedisTemplate())); + assertEquals(0, redisList.size()); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); + MessageChannel redisChannel = context.getBean("listWithKeyAsHeader", MessageChannel.class); + List pepboys = new ArrayList(); + pepboys.add("Manny"); + pepboys.add("Moe"); + pepboys.add("Jack"); + Message> 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 redisList = + new DefaultRedisList("pepboys", + (RedisOperations) this.initTemplate(jcf, new RedisTemplate())); + assertEquals(0, redisList.size()); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); + MessageChannel redisChannel = context.getBean("listWithKeyProvided", MessageChannel.class); + List pepboys = new ArrayList(); + pepboys.add("Manny"); + pepboys.add("Moe"); + pepboys.add("Jack"); + Message> message = MessageBuilder.withPayload(pepboys).build(); + redisChannel.send(message); + + assertEquals(3, redisList.size()); + } + + @SuppressWarnings("unchecked") + @Test + @RedisAvailable + public void testMapToZsetWithProvidedKey(){ + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisZSet redisZset = + new DefaultRedisZSet("presidents", + (RedisOperations) this.initTemplate(jcf, new RedisTemplate())); + assertEquals(0, redisZset.size()); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); + MessageChannel redisChannel = context.getBean("mapToZset", MessageChannel.class); + Map presidents = new HashMap(); + 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> 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 redisMap = + new DefaultRedisMap("pepboys", + (RedisOperations) this.initTemplate(jcf, new RedisTemplate>())); + + assertEquals(0, redisMap.size()); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); + MessageChannel redisChannel = context.getBean("mapToMapA", MessageChannel.class); + Map pepboys = new HashMap(); + pepboys.put("1", "Manny"); + pepboys.put("2", "Moe"); + pepboys.put("3", "Jack"); + + + Message> 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> redisMap = + new DefaultRedisMap>("pepboys", + (RedisOperations) this.initTemplate(jcf, new RedisTemplate>>())); + + assertEquals(0, redisMap.size()); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); + MessageChannel redisChannel = context.getBean("mapToMapB", MessageChannel.class); + Map pepboys = new HashMap(); + pepboys.put("1", "Manny"); + pepboys.put("2", "Moe"); + pepboys.put("3", "Jack"); + + + Message> 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> redisMap = + new DefaultRedisMap>("pepboys", + (RedisOperations) this.initTemplate(jcf, new RedisTemplate>>())); + + assertEquals(0, redisMap.size()); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); + MessageChannel redisChannel = context.getBean("mapToMapB", MessageChannel.class); + Map pepboys = new HashMap(); + pepboys.put("1", "Manny"); + pepboys.put("2", "Moe"); + pepboys.put("3", "Jack"); + + + Message> message = MessageBuilder.withPayload(pepboys).build(); + redisChannel.send(message); + } + + @SuppressWarnings("unchecked") + @Test + @RedisAvailable + public void testMapToMapAsSingleEntryWithKeyAsHeader(){ + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisMap> redisMap = + new DefaultRedisMap>("pepboys", + (RedisOperations) this.initTemplate(jcf, new RedisTemplate>>())); + + assertEquals(0, redisMap.size()); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); + MessageChannel redisChannel = context.getBean("mapToMapB", MessageChannel.class); + Map pepboys = new HashMap(); + pepboys.put("1", "Manny"); + pepboys.put("2", "Moe"); + pepboys.put("3", "Jack"); + + + Message> message = MessageBuilder.withPayload(pepboys). + setHeader(RedisHeaders.KEY, "pepboys").setHeader(RedisHeaders.MAP_KEY, "foo").build(); + redisChannel.send(message); + Map 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 redisList = + new DefaultRedisSet("pepboys", + (RedisOperations) this.initTemplate(jcf, new RedisTemplate())); + assertEquals(0, redisList.size()); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); + MessageChannel redisChannel = context.getBean("set", MessageChannel.class); + Set pepboys = new HashSet(); + pepboys.add("Manny"); + pepboys.add("Moe"); + pepboys.add("Jack"); + Message> 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 redisList = + new DefaultRedisSet("pepboys", + (RedisOperations) this.initTemplate(jcf, new RedisTemplate())); + assertEquals(0, redisList.size()); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); + MessageChannel redisChannel = context.getBean("setNotParsed", MessageChannel.class); + Set pepboys = new HashSet(); + pepboys.add("Manny"); + pepboys.add("Moe"); + pepboys.add("Jack"); + Message> 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 redisList = + new DefaultRedisSet("pepboys", + (RedisOperations) this.initTemplate(jcf, new RedisTemplate())); + 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 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) this.initTemplate(jcf, new RedisTemplate())); + + assertEquals(0, redisProperties.size()); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter.xml", this.getClass()); + MessageChannel redisChannel = context.getBean("property", MessageChannel.class); + Properties pepboys = new Properties(); + pepboys.put("1", "Manny"); + pepboys.put("2", "Moe"); + pepboys.put("3", "Jack"); + + Message 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; + } +} diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-template-cf-fail.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-template-cf-fail.xml index 04592a7bfc..bab98a3fcb 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-template-cf-fail.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-template-cf-fail.xml @@ -9,7 +9,7 @@ diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/store-outbound-adapter.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/store-outbound-adapter.xml new file mode 100644 index 0000000000..9bbbf2d3be --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/store-outbound-adapter.xml @@ -0,0 +1,52 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/zset-inbound-adapter.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/zset-inbound-adapter.xml index cf15ba25d4..88db367d6d 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/zset-inbound-adapter.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/zset-inbound-adapter.xml @@ -45,7 +45,6 @@ - diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisCollectionPopulatingMessageHandlerTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisCollectionPopulatingMessageHandlerTests.java new file mode 100644 index 0000000000..d7aefa520b --- /dev/null +++ b/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 redisList = + new DefaultRedisList(key, + (RedisOperations) this.initTemplate(jcf, new RedisTemplate())); + + assertEquals(0, redisList.size()); + + RedisCollectionPopulatingMessageHandler handler = + new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key)); + + List list = new ArrayList(); + list.add("Manny"); + list.add("Moe"); + list.add("Jack"); + Message> message = new GenericMessage>(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 redisList = + new DefaultRedisList(key, + (RedisOperations) this.initTemplate(jcf, new RedisTemplate())); + + assertEquals(0, redisList.size()); + + RedisCollectionPopulatingMessageHandler handler = + new RedisCollectionPopulatingMessageHandler(jcf, null); + handler.afterPropertiesSet(); + + List list = new ArrayList(); + list.add("Manny"); + list.add("Moe"); + list.add("Jack"); + Message> 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 redisList = + new DefaultRedisList(key, + (RedisOperations) this.initTemplate(jcf, new RedisTemplate())); + + assertEquals(0, redisList.size()); + + RedisCollectionPopulatingMessageHandler handler = + new RedisCollectionPopulatingMessageHandler(jcf, null); + + List list = new ArrayList(); + list.add("Manny"); + list.add("Moe"); + list.add("Jack"); + Message> message = MessageBuilder.withPayload(list).build(); + handler.handleMessage(message); + } + + @SuppressWarnings("unchecked") + @Test + @RedisAvailable + public void testListWithListPayloadAsSingleEntry() { + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + String key = "foo"; + RedisList> redisList = + new DefaultRedisList>(key, + (RedisOperations>) this.initTemplate(jcf, new RedisTemplate>())); + + assertEquals(0, redisList.size()); + + RedisCollectionPopulatingMessageHandler handler = + new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key)); + handler.setExtractPayloadElements(false); + + List list = new ArrayList(); + list.add("Manny"); + list.add("Moe"); + list.add("Jack"); + Message> message = new GenericMessage>(list); + handler.handleMessage(message); + + assertEquals(1, redisList.size()); + List 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 redisZset = + new DefaultRedisZSet(key, + (RedisOperations) this.initTemplate(jcf, new RedisTemplate())); + + assertEquals(0, redisZset.size()); + + RedisCollectionPopulatingMessageHandler handler = + new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key)); + + handler.setCollectionType(CollectionType.ZSET); + + List list = new ArrayList(); + list.add("Manny"); + list.add("Moe"); + list.add("Jack"); + Message> message = new GenericMessage>(list); + handler.handleMessage(message); + + assertEquals(3, redisZset.size()); + Set> pepboys = redisZset.rangeByScoreWithScores(1, 1); + for (TypedTuple pepboy : pepboys) { + assertTrue(pepboy.getScore() == 1); + } + } + + @SuppressWarnings("unchecked") + @Test + @RedisAvailable + public void testZsetWithListPayloadAsSingleEntryAndHeaderKeyHeaderScore() { + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + String key = "foo"; + RedisZSet> redisZset = + new DefaultRedisZSet>(key, + (RedisOperations>) this.initTemplate(jcf, new RedisTemplate>())); + + assertEquals(0, redisZset.size()); + + RedisCollectionPopulatingMessageHandler handler = + new RedisCollectionPopulatingMessageHandler(jcf, null); + handler.afterPropertiesSet(); + + handler.setCollectionType(CollectionType.ZSET); + handler.setExtractPayloadElements(false); + + List list = new ArrayList(); + list.add("Manny"); + list.add("Moe"); + list.add("Jack"); + Message> message = MessageBuilder.withPayload(list).setHeader("redis_key", key). + setHeader("redis_zset_score", 4).build(); + handler.handleMessage(message); + + assertEquals(1, redisZset.size()); + Set>> entries = redisZset.rangeByScoreWithScores(1, 4); + for (TypedTuple> pepboys : entries) { + assertTrue(pepboys.getScore() == 4); + } + } + + @SuppressWarnings("unchecked") + @Test + @RedisAvailable + public void testZsetWithMapPayloadParsedHeaderKey() { + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + String key = "presidents"; + RedisZSet redisZset = + new DefaultRedisZSet(key, + (RedisOperations) this.initTemplate(jcf, new RedisTemplate())); + + assertEquals(0, redisZset.size()); + + RedisCollectionPopulatingMessageHandler handler = + new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key)); + + handler.setCollectionType(CollectionType.ZSET); + + Map presidents = new HashMap(); + 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> message = MessageBuilder.withPayload(presidents).setHeader("redis_key", key).build(); + handler.handleMessage(message); + + assertEquals(13, redisZset.size()); + + Set> entries = redisZset.rangeByScoreWithScores(18, 19); + assertEquals(6, entries.size()); + } + + @SuppressWarnings("unchecked") + @Test + @RedisAvailable + public void testZsetWithMapPayloadPojoParsedHeaderKey() { + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + String key = "presidents"; + RedisZSet redisZset = + new DefaultRedisZSet(key, + (RedisOperations) this.initTemplate(jcf, new RedisTemplate())); + + assertEquals(0, redisZset.size()); + + RedisCollectionPopulatingMessageHandler handler = + new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key)); + + handler.setCollectionType(CollectionType.ZSET); + + Map presidents = new HashMap(); + 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> message = MessageBuilder.withPayload(presidents).setHeader("redis_key", key).build(); + handler.handleMessage(message); + + assertEquals(13, redisZset.size()); + + Set> entries = redisZset.rangeByScoreWithScores(18, 19); + assertEquals(6, entries.size()); + } + + @SuppressWarnings("unchecked") + @Test + @RedisAvailable + public void testZsetWithMapPayloadPojoAsSingleEntryHeaderKey() { + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + String key = "presidents"; + RedisZSet> redisZset = + new DefaultRedisZSet>(key, + (RedisOperations>) this.initTemplate(jcf, new RedisTemplate>())); + + assertEquals(0, redisZset.size()); + + RedisCollectionPopulatingMessageHandler handler = + new RedisCollectionPopulatingMessageHandler(jcf, new LiteralExpression(key)); + + handler.setCollectionType(CollectionType.ZSET); + handler.setExtractPayloadElements(false); + + Map presidents = new HashMap(); + presidents.put(new President("John Adams"), 18D); + + presidents.put(new President("Barack Obama"), 21D); + presidents.put(new President("Thomas Jefferson"), 19D); + + Message> 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; + } + } +}