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

Merge pull request #639 from olegz:INT-2774/75

Browse Source 
# By Oleg Zhurakousky
* olegz-INT-2774/75:
  INT-2774/75 RedisCollectionPopulatingMessageHandler fix * Added support for 'channel' attribute for RedisStore outbound adapter * Fixed logic in RedisCollectionPopulatingMessageHandler to ensure that RedisHeaders.ZSET_INCREMENT_SCORE can be passed as Boolean true/false as well as String true/false. The current solution is based on relyiong oin ConversionService registered with SpEL Context (see RedisCollectionPopulatingMessageHandler.extractZsetIncrementHeader method). This way other types could be supported without introducing any code changes.
This commit is contained in:
Gunnar Hillert
2012-10-03 11:15:50 -04:00
parent 9cd132be3b 3b78490221
commit b85ddc52ed
4 changed files with 76 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -72,6 +72,9 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand
private final Log logger = LogFactory.getLog(this.getClass());
private final Expression zsetIncrementScoreExpression =
new SpelExpressionParser().parseExpression("headers." + RedisHeaders.ZSET_INCREMENT_SCORE);
private volatile StandardEvaluationContext evaluationContext;
private volatile Expression keyExpression =
@@ -322,7 +325,7 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand
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);
final boolean zsetIncrementHeader = this.extractZsetIncrementHeader(message);
if (this.extractPayloadElements) {
@@ -358,6 +361,13 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand
}
}
private boolean extractZsetIncrementHeader(Message<?> message){
if (message.getHeaders().containsKey(RedisHeaders.ZSET_INCREMENT_SCORE)){
return this.zsetIncrementScoreExpression.getValue(this.evaluationContext, message, Boolean.class);
}
return false;
}
@SuppressWarnings("unchecked")
private void handleList(RedisList<Object> list, Message<?> message) {
Object payload = message.getPayload();
@@ -454,15 +464,15 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand
}
private void incrementOrOverwrite(final BoundZSetOperations<String, Object> ops, Object object, Double score,
Object zsetIncrementScore) {
boolean zsetIncrementScore) {
boolean increment = Boolean.TRUE.equals(zsetIncrementScore);;
//boolean increment = Boolean.TRUE.equals(zsetIncrementScore);;
if (score != null) {
this.doIncrementOrOverwrite(ops, object, score, increment);
this.doIncrementOrOverwrite(ops, object, score, zsetIncrementScore);
}
else {
logger.debug("Zset Score could not be determined. Using default score of 1");
this.doIncrementOrOverwrite(ops, object, Double.valueOf(1), increment);
this.doIncrementOrOverwrite(ops, object, Double.valueOf(1), zsetIncrementScore);
}
}

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

@@ -219,15 +219,6 @@
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="topic" type="xsd:string"/>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-converter" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -368,6 +359,18 @@
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Channel to which Messages will be sent.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:complexType name="redisCollectionInboundAdapterType">
@@ -388,18 +391,6 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Channel to which Messages will be sent.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="key-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>

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

@@ -11,6 +11,7 @@
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<int:channel id="someChannel"/>
<int-redis:store-outbound-channel-adapter id="listWithKeyAsHeader"/>
@@ -18,6 +19,11 @@
collection-type="LIST"
key="pepboys"/>
<int-redis:store-outbound-channel-adapter id="listWithKeyProvidedAndChannel"
channel="someChannel"
collection-type="LIST"
key="pepboys"/>
<int-redis:store-outbound-channel-adapter id="mapToZset"
collection-type="ZSET"
key-expression="'presidents'"/>

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

@@ -232,6 +232,49 @@ public class RedisCollectionPopulatingMessageHandlerTests extends RedisAvailable
}
}
@SuppressWarnings("unchecked")
@Test
@RedisAvailable
public void testZsetWithListPayloadParsedAndProvidedKeyScoreIncrementAsStringHeader() {// see INT-2775
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, "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")
@Test
@RedisAvailable