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

INT-3850: Fix SpEL usage in the CacheWritingMH

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

**Cherry-pick to 4.1.x, 4.0.x, 3.0.x**
This commit is contained in:
Artem Bilan
2015-10-09 11:58:01 -04:00
committed by Gary Russell
parent b6467e1329
commit bbfde40694
3 changed files with 87 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 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,37 +21,46 @@ import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;
import com.gemstone.gemfire.GemFireCheckedException;
import com.gemstone.gemfire.GemFireException;
import com.gemstone.gemfire.cache.Region;
import org.springframework.data.gemfire.GemfireCallback;
import org.springframework.data.gemfire.GemfireTemplate;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
import com.gemstone.gemfire.GemFireCheckedException;
import com.gemstone.gemfire.GemFireException;
import com.gemstone.gemfire.cache.Region;
/**
* A {@link MessageHandler} implementation that writes to a GemFire Region. The
* Message's payload must be an instance of java.util.Map.
* Message's payload must be an instance of {@link Map} or {@link #cacheEntryExpressions}
* must be provided.
*
* @author Mark Fisher
* @author David Turanski
* @author Artem Bilan
*
* @since 2.1
*/
public class CacheWritingMessageHandler extends AbstractMessageHandler {
private static final SpelExpressionParser PARSER = new SpelExpressionParser();
private final Map<Expression, Expression> cacheEntryExpressions = new LinkedHashMap<Expression, Expression>();
private final GemfireTemplate gemfireTemplate = new GemfireTemplate();
private volatile EvaluationContext evaluationContext;
@SuppressWarnings("rawtypes")
public CacheWritingMessageHandler(Region region) {
Assert.notNull(region, "region must not be null");
this.gemfireTemplate.setRegion(region);
this.gemfireTemplate.afterPropertiesSet();
}
@Override
@@ -59,52 +68,70 @@ public class CacheWritingMessageHandler extends AbstractMessageHandler {
return "gemfire:outbound-channel-adapter";
}
@SuppressWarnings("unchecked")
@Override
public void handleMessageInternal(Message<?> message) {
protected void onInit() throws Exception {
super.onInit();
this.gemfireTemplate.afterPropertiesSet();
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
}
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
protected void handleMessageInternal(Message<?> message) {
Object payload = message.getPayload();
Map<?, ?> cacheValues = (cacheEntryExpressions.size() > 0) ? parseCacheEntries(message) : null;
Map<?, ?> cacheValues = (this.cacheEntryExpressions.size() > 0) ? evaluateCacheEntries(message) : null;
if (cacheValues == null) {
Assert.isTrue(payload instanceof Map,
Assert.state(payload instanceof Map,
"If cache entry expressions are not configured, then payload must be a Map");
cacheValues = (Map<?, ?>) payload;
cacheValues = (Map) payload;
}
final Map<?, ?> map = cacheValues;
this.gemfireTemplate.execute(new GemfireCallback<Object>() {
@Override
@SuppressWarnings({"rawtypes", "unchecked"})
public Object doInGemfire(Region region) throws GemFireCheckedException, GemFireException {
region.putAll(map);
return null;
}
});
}
private Map<Object, Object> parseCacheEntries(Message<?> message) {
if (cacheEntryExpressions.size() == 0) {
private Map<Object, Object> evaluateCacheEntries(Message<?> message) {
if (this.cacheEntryExpressions.size() == 0) {
return null;
}
else {
Map<Object, Object> cacheValues = new HashMap<Object, Object>();
for (Entry<Expression, Expression> expressionEntry : cacheEntryExpressions.entrySet()) {
cacheValues.put(expressionEntry.getKey().getValue(message), expressionEntry.getValue().getValue(message));
for (Entry<Expression, Expression> expressionEntry : this.cacheEntryExpressions.entrySet()) {
cacheValues.put(expressionEntry.getKey().getValue(this.evaluationContext, message),
expressionEntry.getValue().getValue(this.evaluationContext, message));
}
return cacheValues;
}
}
public void setCacheEntries(Map<String, String> cacheEntries) {
if (cacheEntryExpressions.size() > 0) {
cacheEntryExpressions.clear();
Assert.notNull(cacheEntries, "'cacheEntries' must not be null");
if (this.cacheEntryExpressions.size() > 0) {
this.cacheEntryExpressions.clear();
}
for (Entry<String, String> cacheEntry : cacheEntries.entrySet()) {
this.cacheEntryExpressions.put(new SpelExpressionParser().parseExpression(cacheEntry.getKey()),
new SpelExpressionParser().parseExpression(cacheEntry.getValue()));
this.cacheEntryExpressions.put(PARSER.parseExpression(cacheEntry.getKey()),
PARSER.parseExpression(cacheEntry.getValue()));
}
}
public void setCacheEntryExpressions(Map<Expression, Expression> cacheEntryExpressions) {
Assert.notNull(cacheEntryExpressions, "'cacheEntryExpressions' must not be null");
if (this.cacheEntryExpressions.size() > 0) {
this.cacheEntryExpressions.clear();
}
this.cacheEntryExpressions.putAll(cacheEntryExpressions);
}
}

29
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandlerTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.gemfire.outbound;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock;
import java.util.Collections;
import java.util.HashMap;
@@ -24,10 +25,15 @@ import java.util.Map;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.data.gemfire.CacheFactoryBean;
import org.springframework.data.gemfire.RegionFactoryBean;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import com.gemstone.gemfire.cache.Cache;
import com.gemstone.gemfire.cache.Region;
@@ -37,6 +43,7 @@ import com.gemstone.gemfire.cache.Region;
* @author David Turanski
* @author Gunnar Hillert
* @author Gary Russell
* @author Artem Bilan
* @since 2.1
*/
public class CacheWritingMessageHandlerTests {
@@ -51,7 +58,11 @@ public class CacheWritingMessageHandlerTests {
regionFactoryBean.afterPropertiesSet();
Region<String, String> region = regionFactoryBean.getObject();
assertEquals(0, region.size());
CacheWritingMessageHandler handler = new CacheWritingMessageHandler(region);
handler.setBeanFactory(mock(BeanFactory.class));
handler.afterPropertiesSet();
Map<String, String> map = new HashMap<String, String>();
map.put("foo", "bar");
Message<?> message = MessageBuilder.withPayload(map).build();
@@ -61,14 +72,15 @@ public class CacheWritingMessageHandlerTests {
}
@Test
@SuppressWarnings("unchecked")
public void ExpressionsWriteToCache() throws Exception {
CacheFactoryBean cacheFactoryBean = new CacheFactoryBean();
Cache cache = cacheFactoryBean.getObject();
RegionFactoryBean<String, String> regionFactoryBean = new RegionFactoryBean<String, String>() {};
RegionFactoryBean<String, Object> regionFactoryBean = new RegionFactoryBean<String, Object>() {};
regionFactoryBean.setName("test.expressionsWriteToCache");
regionFactoryBean.setCache(cache);
regionFactoryBean.afterPropertiesSet();
Region<String, String> region = regionFactoryBean.getObject();
Region<String, Object> region = regionFactoryBean.getObject();
assertEquals(0, region.size());
CacheWritingMessageHandler handler = new CacheWritingMessageHandler(region);
@@ -76,12 +88,23 @@ public class CacheWritingMessageHandlerTests {
expressions.put("'foo'", "'bar'");
expressions.put("payload.toUpperCase()", "headers['bar'].toUpperCase()");
handler.setCacheEntries(expressions);
handler.setBeanFactory(mock(BeanFactory.class));
handler.afterPropertiesSet();
Message<?> message = MessageBuilder.withPayload("foo").copyHeaders(Collections.singletonMap("bar", "bar")).build();
Message<?> message = MessageBuilder.withPayload("foo")
.copyHeaders(Collections.singletonMap("bar", "bar"))
.build();
handler.handleMessage(message);
assertEquals(2, region.size());
assertEquals("BAR", region.get("FOO"));
assertEquals("bar", region.get("foo"));
handler.setCacheEntryExpressions(Collections.<Expression, Expression>singletonMap(new LiteralExpression("baz"),
new ValueExpression<Long>(10L)));
handler.handleMessage(new GenericMessage<String>("test"));
assertEquals(3, region.size());
assertEquals(10L, region.get("baz"));
}
}

19
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml
View File

@@ -8,19 +8,24 @@
http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<gfe:cache />
<gfe:replicated-region id="region1"/>
<gfe:replicated-region id="region2"/>
<int-gfe:outbound-channel-adapter id="cacheChannel1" region="region1"/>
<bean id="bar" class="java.lang.String">
<constructor-arg value="bar"/>
</bean>
<int-gfe:outbound-channel-adapter id="cacheChannel2" region="region2" order="19">
<int-gfe:cache-entries>
<entry key="payload.toUpperCase()" value="payload.toLowerCase()"/>
<entry key="'foo'" value="'bar'"/>
</int-gfe:cache-entries>
<entry key="payload.toUpperCase()" value="payload.toLowerCase()"/>
<entry key="'foo'" value="@bar"/>
</int-gfe:cache-entries>
</int-gfe:outbound-channel-adapter>
<int:chain input-channel="cacheChainChannel">