Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #635 from garyrussell/INT-2761

Browse Source 
* INT-2761:
  INT-2761 Make Redis ZSet Default - Increment Score
This commit is contained in:
Oleg Zhurakousky
2012-09-21 12:38:27 -04:00
parent 8ee86f035f eb7638a1a0
commit c8da0532d7
10 changed files with 97 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParser.java → spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionInboundChannelAdapterParser.java
View File

@@ -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) {

2
spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParser.java → spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionOutboundChannelAdapterParser.java
View File

@@ -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) {

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

@@ -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());
}
}

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

@@ -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<Object> zset, final Message<?> message) throws Exception{
final Object payload = message.getPayload();
final BoundZSetOperations<String, Object> ops =
(BoundZSetOperations<String, Object>) this.redisTemplate.boundZSetOps(zset.getKey());
final Object zsetIncrementHeader = message.getHeaders().get(RedisHeaders.ZSET_INCREMENT_SCORE);
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;
final Map<Object, Number> payloadAsMap = (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 ?
for (Entry<Object, Number> 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<Object> zset, Object objectToAdd, Double score) {
private void incrementOrOverwrite(final BoundZSetOperations<String, Object> 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<String, Object> ops, Object object, Double score,
boolean increment) {
if (increment) {
ops.incrementScore(object, score);
}
else {
ops.add(object, score);
}
}

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

@@ -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";
}

2
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterIntegrationTests.java → spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionInboundChannelAdapterIntegrationTests.java
View File

@@ -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

2
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParserTests.java → spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionInboundChannelAdapterParserTests.java
View File

@@ -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());

2
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterIntegrationTests.java → spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionOutboundChannelAdapterIntegrationTests.java
View File

@@ -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

2
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParserTests.java → spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionOutboundChannelAdapterParserTests.java
View File

@@ -28,7 +28,7 @@ import org.springframework.integration.test.util.TestUtils;
*
* @author Oleg Zhurakousky
*/
public class RedisCollectionsOutboundChannelAdapterParserTests {
public class RedisCollectionOutboundChannelAdapterParserTests {
@Test
public void validateFullConfiguration(){

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

@@ -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<String> redisZset =
@@ -178,6 +179,57 @@ public class RedisCollectionPopulatingMessageHandlerTests extends RedisAvailable
for (TypedTuple<String> pepboy : pepboys) {
assertTrue(pepboy.getScore() == 1);
}
handler.handleMessage(message);
assertEquals(3, redisZset.size());
pepboys = redisZset.rangeByScoreWithScores(1, 2);
// should not have incremented
for (TypedTuple<String> pepboy : pepboys) {
assertTrue(pepboy.getScore() == 1);
}
}
@SuppressWarnings("unchecked")
@Test
@RedisAvailable
public void testZsetWithListPayloadParsedAndProvidedKeyScoreIncrement() {
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);
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(RedisHeaders.ZSET_INCREMENT_SCORE, Boolean.TRUE)
.build();
handler.handleMessage(message);
assertEquals(3, redisZset.size());
Set<TypedTuple<String>> pepboys = redisZset.rangeByScoreWithScores(1, 1);
for (TypedTuple<String> pepboy : pepboys) {
assertTrue(pepboy.getScore() == 1);
}
handler.handleMessage(message);
assertEquals(3, redisZset.size());
pepboys = redisZset.rangeByScoreWithScores(1, 2);
// should have incremented
for (TypedTuple<String> pepboy : pepboys) {
assertTrue(pepboy.getScore() == 2);
}
}
@SuppressWarnings("unchecked")
@@ -204,7 +256,7 @@ public class RedisCollectionPopulatingMessageHandlerTests extends RedisAvailable
list.add("Moe");
list.add("Jack");
Message<List<String>> 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());