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

INT-4258: Redis: Make ZADD INCR False by Default

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4258

Since Redis `ZADD` command doesn't perform incrementation by default,
it should be consistent in the `RedisStoreWritingMessageHandler` as well

* Make a default behavior of the `RedisStoreWritingMessageHandler`
regarding `INCR` option as `false`
* Provide some performance refactoring for the
`RedisStoreWritingMessageHandler`  moving the code to Java 8 style
and using `FunctionExpression`
* Fix tests according to the changed default behavior

* Add `setZsetIncrementExpression` and String-based setters for other
expressions on the `RedisStoreWritingMessageHandler`
* Add `zset-increment-expression` option for XML configuration
* Add `What's New` note on the matter

Polishing
This commit is contained in:
Artem Bilan
2017-05-03 16:26:15 -04:00
committed by Gary Russell
parent 2dfb0808b3
commit 47433754b8
8 changed files with 151 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisStoreOutboundChannelAdapterParser.java
View File

@@ -21,9 +21,7 @@ import org.w3c.dom.Element;
import org.springframework.beans.factory.config.TypedStringValue;
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.RedisStoreWritingMessageHandler;
@@ -35,13 +33,16 @@ import org.springframework.util.StringUtils;
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Mark Fisher
* @author Artem Bilan
*
* @since 2.2
*/
public class RedisStoreOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisStoreWritingMessageHandler.class);
BeanDefinitionBuilder builder =
BeanDefinitionBuilder.genericBeanDefinition(RedisStoreWritingMessageHandler.class);
String redisTemplateRef = element.getAttribute("redis-template");
String connectionFactory = element.getAttribute("connection-factory");
@@ -69,20 +70,18 @@ public class RedisStoreOutboundChannelAdapterParser extends AbstractOutboundChan
builder.addPropertyValue("key", new TypedStringValue(element.getAttribute("key")));
}
if (hasKeyExpression) {
RootBeanDefinition expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class);
expressionDef.getConstructorArgumentValues().addGenericArgumentValue(element.getAttribute("key-expression"));
builder.addPropertyValue("keyExpression", expressionDef);
builder.addPropertyValue("keyExpressionString", element.getAttribute("key-expression"));
}
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);
builder.addPropertyValue("mapKeyExpressionString", mapKeyExpression);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "collection-type");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload-elements");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "zset-increment-expression",
"zsetIncrementExpressionString");
return builder.getBeanDefinition();
}

174
spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisStoreWritingMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2007-2016 the original author or authors.
* Copyright 2007-2017 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.
@@ -21,9 +21,6 @@ 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;
@@ -43,9 +40,9 @@ 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.common.LiteralExpression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.redis.support.RedisHeaders;
import org.springframework.messaging.Message;
@@ -72,34 +69,37 @@ import org.springframework.util.NumberUtils;
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Mark Fisher
* @author Artem Bilan
*
* @since 2.2
*/
public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
private final Log logger = LogFactory.getLog(this.getClass());
private Expression zsetIncrementScoreExpression =
new FunctionExpression<Message<?>>(m ->
m.getHeaders().get(RedisHeaders.ZSET_INCREMENT_SCORE));
private final Expression zsetIncrementScoreExpression =
new SpelExpressionParser().parseExpression("headers." + RedisHeaders.ZSET_INCREMENT_SCORE);
private Expression keyExpression =
new FunctionExpression<Message<?>>(m ->
m.getHeaders().get(RedisHeaders.KEY));
private volatile StandardEvaluationContext evaluationContext;
private Expression mapKeyExpression =
new FunctionExpression<Message<?>>(m ->
m.getHeaders().get(RedisHeaders.MAP_KEY));
private volatile Expression keyExpression =
new SpelExpressionParser().parseExpression("headers." + RedisHeaders.KEY);
private boolean mapKeyExpressionExplicitlySet;
private volatile Expression mapKeyExpression =
new SpelExpressionParser().parseExpression("headers." + RedisHeaders.MAP_KEY);
private StandardEvaluationContext evaluationContext;
private volatile boolean mapKeyExpressionExplicitlySet;
private RedisTemplate<String, ?> redisTemplate = new StringRedisTemplate();
private volatile RedisTemplate<String, ?> redisTemplate = new StringRedisTemplate();
private boolean redisTemplateExplicitlySet;
private volatile boolean redisTemplateExplicitlySet;
private CollectionType collectionType = CollectionType.LIST;
private volatile CollectionType collectionType = CollectionType.LIST;
private boolean extractPayloadElements = true;
private volatile boolean extractPayloadElements = true;
private volatile RedisConnectionFactory connectionFactory;
private RedisConnectionFactory connectionFactory;
private volatile boolean initialized;
@@ -107,7 +107,6 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
/**
* Constructs an instance using the provided {@link RedisTemplate}.
* The RedisTemplate must be fully initialized.
*
* @param redisTemplate The Redis template.
*/
public RedisStoreWritingMessageHandler(RedisTemplate<String, ?> redisTemplate) {
@@ -122,9 +121,8 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
* true (default) or a {@link RedisTemplate} with {@link StringRedisSerializer}s for
* keys and hash keys and {@link JdkSerializationRedisSerializer}s for values and
* hash values, when it is false.
*
* @see #setExtractPayloadElements(boolean)
* @param connectionFactory The connection factory.
* @see #setExtractPayloadElements(boolean)
*/
public RedisStoreWritingMessageHandler(RedisConnectionFactory connectionFactory) {
Assert.notNull(connectionFactory, "'connectionFactory' must not be null");
@@ -136,9 +134,8 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
* Specifies the key for the Redis store. If an expression is needed, then call
* {@link #setKeyExpression(Expression)} instead of this method (they are mutually exclusive).
* If neither setter is called, the default expression will be 'headers.{@link RedisHeaders#KEY}'.
*
* @see #setKeyExpression(Expression)
* @param key The key.
* @see #setKeyExpression
*/
public void setKey(String key) {
Assert.hasText(key, "key must not be empty");
@@ -150,9 +147,22 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
* If an expression is not needed, then a literal value may be passed to the
* {@link #setKey(String)} method instead of this one (they are mutually exclusive).
* If neither setter is called, the default expression will be 'headers.{@link RedisHeaders#KEY}'.
*
* @see #setKey(String)
* @param keyExpression The key expression.
* @see #setKey(String)
* @since 5.0
*/
public void setKeyExpressionString(String keyExpression) {
Assert.hasText(keyExpression, "'keyExpression' must not be empty");
setKeyExpression(EXPRESSION_PARSER.parseExpression(keyExpression));
}
/**
* Specifies a SpEL Expression to be used to determine the key for the Redis store.
* If an expression is not needed, then a literal value may be passed to the
* {@link #setKey(String)} method instead of this one (they are mutually exclusive).
* If neither setter is called, the default expression will be 'headers.{@link RedisHeaders#KEY}'.
* @param keyExpression The key expression.
* @see #setKey(String)
*/
public void setKeyExpression(Expression keyExpression) {
Assert.notNull(keyExpression, "keyExpression must not be null");
@@ -161,7 +171,6 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
/**
* Sets the collection type for this handler as per {@link CollectionType}.
*
* @param collectionType The collection type.
*/
public void setCollectionType(CollectionType collectionType) {
@@ -175,7 +184,6 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
* 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.
*
* @param extractPayloadElements true if payload elements should be extracted.
*/
public void setExtractPayloadElements(boolean extractPayloadElements) {
@@ -185,7 +193,17 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
/**
* Sets the expression used as the key for Map and Properties entries.
* Default is 'headers.{@link RedisHeaders#MAP_KEY}'
*
* @param mapKeyExpression The map key expression.
* @since 5.0
*/
public void setMapKeyExpressionString(String mapKeyExpression) {
Assert.hasText(mapKeyExpression, "'mapKeyExpression' must not be empty");
setMapKeyExpression(EXPRESSION_PARSER.parseExpression(mapKeyExpression));
}
/**
* Sets the expression used as the key for Map and Properties entries.
* Default is 'headers.{@link RedisHeaders#MAP_KEY}'
* @param mapKeyExpression The map key expression.
*/
public void setMapKeyExpression(Expression mapKeyExpression) {
@@ -194,6 +212,28 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
this.mapKeyExpressionExplicitlySet = true;
}
/**
* Set the expression used as the INCR flag for the ZADD command in case of ZSet collection.
* Default is 'headers.{@link RedisHeaders#ZSET_INCREMENT_SCORE}'
* @param zsetIncrementScoreExpression The ZADD INCR flag expression.
* @since 5.0
*/
public void setZsetIncrementExpressionString(String zsetIncrementScoreExpression) {
Assert.hasText(zsetIncrementScoreExpression, "'zsetIncrementScoreExpression' must not be empty");
setZsetIncrementExpression(EXPRESSION_PARSER.parseExpression(zsetIncrementScoreExpression));
}
/**
* Set the expression used as the INCR flag for the ZADD command in case of ZSet collection.
* Default is 'headers.{@link RedisHeaders#ZSET_INCREMENT_SCORE}'
* @param zsetIncrementScoreExpression The ZADD INCR flag expression.
* @since 5.0
*/
public void setZsetIncrementExpression(Expression zsetIncrementScoreExpression) {
Assert.notNull(zsetIncrementScoreExpression, "'zsetIncrementScoreExpression' must not be null");
this.zsetIncrementScoreExpression = zsetIncrementScoreExpression;
}
@Override
public String getComponentType() {
return "redis:store-outbound-channel-adapter";
@@ -202,13 +242,13 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
@Override
protected void onInit() throws Exception {
this.evaluationContext =
ExpressionUtils.createStandardEvaluationContext(this.getBeanFactory());
ExpressionUtils.createStandardEvaluationContext(this.getBeanFactory());
Assert.state(!this.mapKeyExpressionExplicitlySet ||
(this.collectionType == CollectionType.MAP || this.collectionType == CollectionType.PROPERTIES),
(this.collectionType == CollectionType.MAP || this.collectionType == CollectionType.PROPERTIES),
"'mapKeyExpression' can only be set for CollectionType.MAP or CollectionType.PROPERTIES");
if (!this.redisTemplateExplicitlySet) {
if (!this.extractPayloadElements) {
RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
RedisTemplate<String, Object> template = new RedisTemplate<>();
StringRedisSerializer serializer = new StringRedisSerializer();
template.setKeySerializer(serializer);
template.setHashKeySerializer(serializer);
@@ -251,27 +291,27 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
String key = this.keyExpression.getValue(this.evaluationContext, message, String.class);
Assert.hasText(key, "Failed to determine a key for the Redis store using expression: "
+ this.keyExpression.getExpressionString());
Assert.hasText(key, () -> "Failed to determine a key for the Redis store based on the message: " + message);
RedisStore store = this.createStoreView(key);
Assert.state(this.initialized, "handler not initialized - afterPropertiesSet() must be called before the first use");
Assert.state(this.initialized,
"handler not initialized - afterPropertiesSet() must be called before the first use");
try {
if (this.collectionType == CollectionType.ZSET) {
this.writeToZset((RedisZSet<Object>) store, message);
writeToZset((RedisZSet<Object>) store, message);
}
else if (this.collectionType == CollectionType.SET) {
this.writeToSet((RedisSet<Object>) store, message);
writeToSet((RedisSet<Object>) store, message);
}
else if (this.collectionType == CollectionType.LIST) {
this.writeToList((RedisList<Object>) store, message);
writeToList((RedisList<Object>) store, message);
}
else if (this.collectionType == CollectionType.MAP) {
this.writeToMap((RedisMap<Object, Object>) store, message);
writeToMap((RedisMap<Object, Object>) store, message);
}
else if (this.collectionType == CollectionType.PROPERTIES) {
this.writeToProperties((RedisProperties) store, message);
writeToProperties((RedisProperties) store, message);
}
}
catch (Exception e) {
@@ -284,49 +324,46 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
final Object payload = message.getPayload();
final BoundZSetOperations<String, Object> ops =
(BoundZSetOperations<String, Object>) this.redisTemplate.boundZSetOps(zset.getKey());
final boolean zsetIncrementHeader = this.extractZsetIncrementHeader(message);
boolean zsetIncrementHeader = extractZsetIncrementHeader(message);
if (this.extractPayloadElements) {
if ((payload instanceof Map<?, ?> && this.verifyAllMapValuesOfTypeNumber((Map<?, ?>) payload))) {
final Map<Object, Number> payloadAsMap = (Map<Object, Number>) payload;
this.processInPipeline(() -> {
Map<Object, Number> payloadAsMap = (Map<Object, Number>) payload;
processInPipeline(() -> {
for (Entry<Object, Number> entry : payloadAsMap.entrySet()) {
Number d = entry.getValue();
incrementOrOverwrite(ops, entry.getKey(), d == null ?
determineScore(message) :
NumberUtils.convertNumberToTargetClass(d, Double.class),
determineScore(message) :
NumberUtils.convertNumberToTargetClass(d, Double.class),
zsetIncrementHeader);
}
});
}
else if (payload instanceof Collection<?>) {
this.processInPipeline(() -> {
processInPipeline(() -> {
for (Object object : ((Collection<?>) payload)) {
incrementOrOverwrite(ops, object, determineScore(message), zsetIncrementHeader);
}
});
}
else {
this.incrementOrOverwrite(ops, payload, this.determineScore(message), zsetIncrementHeader);
incrementOrOverwrite(ops, payload, this.determineScore(message), zsetIncrementHeader);
}
}
else {
this.incrementOrOverwrite(ops, payload, this.determineScore(message), zsetIncrementHeader);
incrementOrOverwrite(ops, payload, this.determineScore(message), zsetIncrementHeader);
}
}
private boolean extractZsetIncrementHeader(Message<?> message) {
if (message.getHeaders().containsKey(RedisHeaders.ZSET_INCREMENT_SCORE)) {
return this.zsetIncrementScoreExpression.getValue(this.evaluationContext, message, Boolean.class);
}
return true;
Boolean value = this.zsetIncrementScoreExpression.getValue(this.evaluationContext, message, Boolean.class);
return value != null ? value : false;
}
@SuppressWarnings("unchecked")
private void writeToList(RedisList<Object> list, Message<?> message) {
Object payload = message.getPayload();
if (this.extractPayloadElements) {
if (payload instanceof Collection<?>) {
list.addAll((Collection<? extends Object>) payload);
list.addAll((Collection<?>) payload);
}
else {
list.add(payload);
@@ -341,10 +378,10 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
private void writeToSet(final RedisSet<Object> set, Message<?> message) {
final Object payload = message.getPayload();
if (this.extractPayloadElements && payload instanceof Collection<?>) {
final BoundSetOperations<String, Object> ops =
BoundSetOperations<String, Object> ops =
(BoundSetOperations<String, Object>) this.redisTemplate.boundSetOps(set.getKey());
this.processInPipeline(() -> {
processInPipeline(() -> {
for (Object object : ((Collection<?>) payload)) {
ops.add(object);
}
@@ -355,11 +392,10 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
}
}
@SuppressWarnings("unchecked")
private void writeToMap(final RedisMap<Object, Object> map, Message<?> message) {
final Object payload = message.getPayload();
if (this.extractPayloadElements && payload instanceof Map<?, ?>) {
this.processInPipeline(() -> map.putAll((Map<? extends Object, ? extends Object>) payload));
processInPipeline(() -> map.putAll((Map<?, ?>) payload));
}
else {
Object key = this.determineMapKey(message, false);
@@ -370,7 +406,7 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
private void writeToProperties(final RedisProperties properties, Message<?> message) {
final Object payload = message.getPayload();
if (this.extractPayloadElements && payload instanceof Properties) {
this.processInPipeline(() -> properties.putAll((Properties) payload));
processInPipeline(() -> properties.putAll((Properties) payload));
}
else {
Assert.isInstanceOf(String.class, payload, "For property, payload must be a String.");
@@ -394,29 +430,25 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
private Object determineMapKey(Message<?> message, boolean property) {
Object mapKey = this.mapKeyExpression.getValue(this.evaluationContext, message);
Assert.notNull(mapKey, "Cannot determine a map key for the entry. The key is determined by evaluating " +
"the 'mapKeyExpression' property.");
Assert.notNull(mapKey, () -> "Cannot determine a map key for the entry based on the message: " + message);
if (property) {
Assert.isInstanceOf(String.class, mapKey, "For property, key 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 incrementOrOverwrite(final BoundZSetOperations<String, Object> ops, Object object, Double score,
private void incrementOrOverwrite(BoundZSetOperations<String, Object> ops, Object object, Double score,
boolean zsetIncrementScore) {
if (score != null) {
this.doIncrementOrOverwrite(ops, object, score, zsetIncrementScore);
doIncrementOrOverwrite(ops, object, score, zsetIncrementScore);
}
else {
this.logger.debug("Zset Score could not be determined. Using default score of 1");
this.doIncrementOrOverwrite(ops, object, Double.valueOf(1), zsetIncrementScore);
doIncrementOrOverwrite(ops, object, 1d, zsetIncrementScore);
}
}
private void doIncrementOrOverwrite(final BoundZSetOperations<String, Object> ops, Object object, Double score,
private void doIncrementOrOverwrite(BoundZSetOperations<String, Object> ops, Object object, Double score,
boolean increment) {
if (increment) {
ops.incrementScore(object, score);
@@ -451,7 +483,7 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
private double determineScore(Message<?> message) {
Object scoreHeader = message.getHeaders().get(RedisHeaders.ZSET_SCORE);
if (scoreHeader == null) {
return Double.valueOf(1);
return 1d;
}
else {
Assert.isInstanceOf(Number.class, scoreHeader, "Header " + RedisHeaders.ZSET_SCORE + " must be a Number");
@@ -461,7 +493,9 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
}
private interface PipelineCallback {
void process();
}
}

8
spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-5.0.xsd
View File

@@ -1080,6 +1080,14 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="zset-increment-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
SpEL expression that returns the INCR flag for the ZADD command. Only applies
if the 'collection-type' is ZSET.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="extract-payload-elements" type="xsd:string">
<xsd:annotation>
<xsd:documentation>

1
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisStoreOutboundChannelAdapterParserTests-context.xml
View File

@@ -22,6 +22,7 @@
<int-redis:store-outbound-channel-adapter id="withStringTemplate"
collection-type="${collection.type}"
key="pepboys"
zset-increment-expression="true"
auto-startup="false">
<int-redis:request-handler-advice-chain>
<int:retry-advice/>

9
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisStoreOutboundChannelAdapterParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2007-2016 the original author or authors.
* Copyright 2007-2017 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.
@@ -34,6 +34,7 @@ import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice;
import org.springframework.integration.redis.outbound.RedisStoreWritingMessageHandler;
@@ -41,6 +42,7 @@ import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
*
* @author Oleg Zhurakousky
@@ -61,7 +63,7 @@ public class RedisStoreOutboundChannelAdapterParserTests {
@Test
public void validateWithStringTemplate() throws Exception {
RedisStoreWritingMessageHandler withStringTemplate = context.getBean("withStringTemplate.handler",
RedisStoreWritingMessageHandler.class);
RedisStoreWritingMessageHandler.class);
assertEquals("pepboys", ((LiteralExpression) TestUtils.getPropertyValue(withStringTemplate,
"keyExpression")).getExpressionString());
assertEquals("PROPERTIES", (TestUtils.getPropertyValue(withStringTemplate, "collectionType")).toString());
@@ -75,6 +77,9 @@ public class RedisStoreOutboundChannelAdapterParserTests {
assertThat(TestUtils.getPropertyValue(handler, "h.advised.advisors.first.item.advice"),
Matchers.instanceOf(RequestHandlerRetryAdvice.class));
assertEquals("true", TestUtils.getPropertyValue(withStringTemplate, "zsetIncrementScoreExpression",
Expression.class).getExpressionString());
}
@Test

15
spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreOutboundChannelAdapterIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2007-2016 the original author or authors.
* Copyright 2007-2017 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.
@@ -66,6 +66,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.2
*/
@ContextConfiguration
@@ -190,7 +191,10 @@ public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvail
RedisZSet<String> redisZSet = new DefaultRedisZSet<String>("foo", this.redisTemplate);
assertEquals(0, redisZSet.size());
Message<String> message = MessageBuilder.withPayload("bar").setHeader(RedisHeaders.KEY, "foo").build();
Message<String> message = MessageBuilder.withPayload("bar")
.setHeader(RedisHeaders.KEY, "foo")
.setHeader(RedisHeaders.ZSET_INCREMENT_SCORE, true)
.build();
this.zsetChannel.send(message);
assertEquals(1, redisZSet.size());
@@ -210,7 +214,6 @@ public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvail
Message<String> message = MessageBuilder.withPayload("bar")
.setHeader(RedisHeaders.KEY, "foo")
.setHeader(RedisHeaders.ZSET_INCREMENT_SCORE, false)
.build();
this.zsetChannel.send(message);
@@ -232,6 +235,7 @@ public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvail
Message<String> message = MessageBuilder.withPayload("bar")
.setHeader(RedisHeaders.KEY, "foo")
.setHeader(RedisHeaders.ZSET_SCORE, 2)
.setHeader(RedisHeaders.ZSET_INCREMENT_SCORE, true)
.build();
this.zsetChannel.send(message);
@@ -280,7 +284,9 @@ public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvail
presidents.put("John Quincy Adams", 19);
presidents.put("Zachary Taylor", 19);
Message<Map<String, Integer>> message = MessageBuilder.withPayload(presidents).build();
Message<Map<String, Integer>> message = MessageBuilder.withPayload(presidents)
.setHeader(RedisHeaders.ZSET_INCREMENT_SCORE, true)
.build();
this.mapToZsetChannel.send(message);
@@ -293,7 +299,6 @@ public class RedisStoreOutboundChannelAdapterIntegrationTests extends RedisAvail
this.beanFactory.getBean("mapToZset.handler", RedisStoreWritingMessageHandler.class);
assertEquals("'presidents'", TestUtils.getPropertyValue(handler, "keyExpression.expression"));
// test default (increment by score) behavior
this.mapToZsetChannel.send(message);
assertEquals(5, redisZset.size());

7
spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisStoreWritingMessageHandlerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -55,6 +55,7 @@ import org.springframework.messaging.support.GenericMessage;
* @author Gunnar Hillert
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*/
public class RedisStoreWritingMessageHandlerTests extends RedisAvailableTests {
@@ -201,7 +202,9 @@ public class RedisStoreWritingMessageHandlerTests extends RedisAvailableTests {
list.add("Manny");
list.add("Moe");
list.add("Jack");
Message<List<String>> message = new GenericMessage<List<String>>(list);
Message<List<String>> message = MessageBuilder.withPayload(list)
.setHeader(RedisHeaders.ZSET_INCREMENT_SCORE, true)
.build();
handler.handleMessage(message);
assertEquals(3, redisZset.size());