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/RedisCollectionInboundChannelAdapterParser.java similarity index 96% rename from spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParser.java rename to spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionInboundChannelAdapterParser.java index 3a7dbef971..3497e7457c 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/RedisCollectionInboundChannelAdapterParser.java @@ -33,7 +33,7 @@ import org.springframework.util.StringUtils; * @author Oleg Zhurakousky * @since 2.2 */ -public class RedisCollectionsInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser { +public class RedisCollectionInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser { @Override protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) { 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/RedisCollectionOutboundChannelAdapterParser.java similarity index 97% rename from spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParser.java rename to spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionOutboundChannelAdapterParser.java index cb8a131be8..1d2bc76d14 100644 --- 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/RedisCollectionOutboundChannelAdapterParser.java @@ -32,7 +32,7 @@ import org.springframework.util.StringUtils; * @author Oleg Zhurakousky * @since 2.2 */ -public class RedisCollectionsOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser { +public class RedisCollectionOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser { @Override protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { 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 749ce9f1e1..ef82cbefa8 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 @@ -28,8 +28,8 @@ public class RedisNamespaceHandler extends AbstractIntegrationNamespaceHandler { public void init() { 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("store-inbound-channel-adapter", new RedisCollectionInboundChannelAdapterParser()); + registerBeanDefinitionParser("store-outbound-channel-adapter", new RedisCollectionOutboundChannelAdapterParser()); 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 index 7491043fe7..7d63aa7ca0 100644 --- 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 @@ -17,10 +17,12 @@ package org.springframework.integration.redis.outbound; import java.util.Collection; import java.util.Map; +import java.util.Map.Entry; 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; @@ -318,20 +320,22 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand @SuppressWarnings("unchecked") private void handleZset(RedisZSet zset, final Message message) throws Exception{ final Object payload = message.getPayload(); + final BoundZSetOperations ops = + (BoundZSetOperations) this.redisTemplate.boundZSetOps(zset.getKey()); + final Object zsetIncrementHeader = message.getHeaders().get(RedisHeaders.ZSET_INCREMENT_SCORE); 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; + final Map payloadAsMap = (Map) payload; this.processInPipeline(new PipelineCallback() { public void process() { - for (Object key : pyloadAsMap.keySet()) { - Number d = pyloadAsMap.get(key); - ops.add(key, d == null ? + for (Entry entry : payloadAsMap.entrySet()) { + Number d = entry.getValue(); + incrementOrOverwrite(ops, entry.getKey(), d == null ? determineScore(message) : - NumberUtils.convertNumberToTargetClass(d, Double.class)); + NumberUtils.convertNumberToTargetClass(d, Double.class), + zsetIncrementHeader); } } }); @@ -340,17 +344,17 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand this.processInPipeline(new PipelineCallback() { public void process() { for (Object object : ((Collection)payload)) { - ops.add(object, determineScore(message)); + incrementOrOverwrite(ops, object, determineScore(message), zsetIncrementHeader); } } }); } else { - this.addToZset(zset, payload, this.determineScore(message)); + this.incrementOrOverwrite(ops, payload, this.determineScore(message), zsetIncrementHeader); } } else { - this.addToZset(zset, payload, this.determineScore(message)); + this.incrementOrOverwrite(ops, payload, this.determineScore(message), zsetIncrementHeader); } } @@ -449,13 +453,26 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand return mapKey; } - private void addToZset(RedisZSet zset, Object objectToAdd, Double score) { + private void incrementOrOverwrite(final BoundZSetOperations ops, Object object, Double score, + Object zsetIncrementScore) { + + boolean increment = Boolean.TRUE.equals(zsetIncrementScore);; if (score != null) { - zset.add(objectToAdd, score); + this.doIncrementOrOverwrite(ops, object, score, increment); } else { logger.debug("Zset Score could not be determined. Using default score of 1"); - zset.add(objectToAdd); + this.doIncrementOrOverwrite(ops, object, Double.valueOf(1), increment); + } + } + + private void doIncrementOrOverwrite(final BoundZSetOperations ops, Object object, Double score, + boolean increment) { + if (increment) { + ops.incrementScore(object, score); + } + else { + ops.add(object, score); } } 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 index 7fbc664a51..5f428c3813 100644 --- 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 @@ -5,6 +5,7 @@ package org.springframework.integration.redis.support; * for dealing with headers required by Redis components * * @author Oleg Zhurakousky + * @author Gary Russell * @since 2.2 */ public class RedisHeaders { @@ -13,8 +14,10 @@ public class RedisHeaders { public static final String KEY = PREFIX + "key"; - public static final String MAP_KEY = PREFIX + "map_key"; + public static final String MAP_KEY = PREFIX + "mapKey"; - public static final String ZSET_SCORE = PREFIX + "zset_score"; + public static final String ZSET_SCORE = PREFIX + "zsetScore"; + + public static final String ZSET_INCREMENT_SCORE = PREFIX + "zsetIncrementScore"; } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterIntegrationTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionInboundChannelAdapterIntegrationTests.java similarity index 99% rename from spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterIntegrationTests.java rename to spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionInboundChannelAdapterIntegrationTests.java index ba3ecdee03..8bdfa03c29 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterIntegrationTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionInboundChannelAdapterIntegrationTests.java @@ -36,7 +36,7 @@ import org.springframework.integration.redis.rules.RedisAvailableTests; * @author Oleg Zhurakousky * @since 2.2 */ -public class RedisCollectionsInboundChannelAdapterIntegrationTests extends RedisAvailableTests{ +public class RedisCollectionInboundChannelAdapterIntegrationTests extends RedisAvailableTests{ @Test @RedisAvailable diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionInboundChannelAdapterParserTests.java similarity index 97% rename from spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParserTests.java rename to spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionInboundChannelAdapterParserTests.java index ff242dd74d..94bbaa9da4 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionInboundChannelAdapterParserTests.java @@ -30,7 +30,7 @@ import org.springframework.integration.test.util.TestUtils; * @author Oleg Zhurakousky * @since 2.2 */ -public class RedisCollectionsInboundChannelAdapterParserTests { +public class RedisCollectionInboundChannelAdapterParserTests { @Test public void validateFullConfiguration(){ ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("inbound-store-adapter-parser.xml", this.getClass()); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterIntegrationTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionOutboundChannelAdapterIntegrationTests.java similarity index 99% rename from spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterIntegrationTests.java rename to spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionOutboundChannelAdapterIntegrationTests.java index bff61d49fe..6c3fa9a920 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterIntegrationTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionOutboundChannelAdapterIntegrationTests.java @@ -57,7 +57,7 @@ import org.springframework.integration.test.util.TestUtils; * @author Oleg Zhurakousky * @since 2.2 */ -public class RedisCollectionsOutboundChannelAdapterIntegrationTests extends RedisAvailableTests { +public class RedisCollectionOutboundChannelAdapterIntegrationTests extends RedisAvailableTests { @SuppressWarnings("unchecked") @Test 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/RedisCollectionOutboundChannelAdapterParserTests.java similarity index 97% rename from spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParserTests.java rename to spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionOutboundChannelAdapterParserTests.java index 2fd68d12bb..90ce9966e7 100644 --- 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/RedisCollectionOutboundChannelAdapterParserTests.java @@ -28,7 +28,7 @@ import org.springframework.integration.test.util.TestUtils; * * @author Oleg Zhurakousky */ -public class RedisCollectionsOutboundChannelAdapterParserTests { +public class RedisCollectionOutboundChannelAdapterParserTests { @Test public void validateFullConfiguration(){ 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 index 0a31338691..e90a3658a4 100644 --- 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 @@ -29,6 +29,7 @@ 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.redis.support.RedisHeaders; import org.springframework.integration.support.MessageBuilder; public class RedisCollectionPopulatingMessageHandlerTests extends RedisAvailableTests{ @@ -151,7 +152,7 @@ public class RedisCollectionPopulatingMessageHandlerTests extends RedisAvailable @SuppressWarnings("unchecked") @Test @RedisAvailable - public void testZsetWithListPayloadParsedAndProvidedKeyDefaultScore() { + public void testZsetWithListPayloadParsedAndProvidedKeyDefault() { JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); String key = "foo"; RedisZSet redisZset = @@ -178,6 +179,57 @@ public class RedisCollectionPopulatingMessageHandlerTests extends RedisAvailable for (TypedTuple pepboy : pepboys) { assertTrue(pepboy.getScore() == 1); } + + handler.handleMessage(message); + assertEquals(3, redisZset.size()); + pepboys = redisZset.rangeByScoreWithScores(1, 2); + // should not have incremented + for (TypedTuple pepboy : pepboys) { + assertTrue(pepboy.getScore() == 1); + } + } + + @SuppressWarnings("unchecked") + @Test + @RedisAvailable + public void testZsetWithListPayloadParsedAndProvidedKeyScoreIncrement() { + 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); + handler.afterPropertiesSet(); + + List list = new ArrayList(); + list.add("Manny"); + list.add("Moe"); + list.add("Jack"); + Message> message = MessageBuilder.withPayload(list) + .setHeader(RedisHeaders.ZSET_INCREMENT_SCORE, Boolean.TRUE) + .build(); + + handler.handleMessage(message); + + assertEquals(3, redisZset.size()); + Set> pepboys = redisZset.rangeByScoreWithScores(1, 1); + for (TypedTuple pepboy : pepboys) { + assertTrue(pepboy.getScore() == 1); + } + + handler.handleMessage(message); + assertEquals(3, redisZset.size()); + pepboys = redisZset.rangeByScoreWithScores(1, 2); + // should have incremented + for (TypedTuple pepboy : pepboys) { + assertTrue(pepboy.getScore() == 2); + } } @SuppressWarnings("unchecked") @@ -204,7 +256,7 @@ public class RedisCollectionPopulatingMessageHandlerTests extends RedisAvailable list.add("Moe"); list.add("Jack"); Message> message = MessageBuilder.withPayload(list).setHeader("redis_key", key). - setHeader("redis_zset_score", 4).build(); + setHeader("redis_zsetScore", 4).build(); handler.handleMessage(message); assertEquals(1, redisZset.size());