diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandler.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandler.java index 5641c9d766..9e164f40ac 100644 --- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandler.java +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandler.java @@ -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 cacheEntryExpressions = new LinkedHashMap(); 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() { + @Override - @SuppressWarnings({"rawtypes", "unchecked"}) public Object doInGemfire(Region region) throws GemFireCheckedException, GemFireException { region.putAll(map); return null; } + }); } - private Map parseCacheEntries(Message message) { - if (cacheEntryExpressions.size() == 0) { + private Map evaluateCacheEntries(Message message) { + if (this.cacheEntryExpressions.size() == 0) { return null; } else { Map cacheValues = new HashMap(); - for (Entry expressionEntry : cacheEntryExpressions.entrySet()) { - cacheValues.put(expressionEntry.getKey().getValue(message), expressionEntry.getValue().getValue(message)); + for (Entry expressionEntry : this.cacheEntryExpressions.entrySet()) { + cacheValues.put(expressionEntry.getKey().getValue(this.evaluationContext, message), + expressionEntry.getValue().getValue(this.evaluationContext, message)); } return cacheValues; } } public void setCacheEntries(Map 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 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 cacheEntryExpressions) { + Assert.notNull(cacheEntryExpressions, "'cacheEntryExpressions' must not be null"); + if (this.cacheEntryExpressions.size() > 0) { + this.cacheEntryExpressions.clear(); + } + this.cacheEntryExpressions.putAll(cacheEntryExpressions); + } + } diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandlerTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandlerTests.java index 8251a50cf9..489c95da25 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandlerTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandlerTests.java @@ -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 region = regionFactoryBean.getObject(); assertEquals(0, region.size()); + CacheWritingMessageHandler handler = new CacheWritingMessageHandler(region); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.afterPropertiesSet(); + Map map = new HashMap(); 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 regionFactoryBean = new RegionFactoryBean() {}; + RegionFactoryBean regionFactoryBean = new RegionFactoryBean() {}; regionFactoryBean.setName("test.expressionsWriteToCache"); regionFactoryBean.setCache(cache); regionFactoryBean.afterPropertiesSet(); - Region region = regionFactoryBean.getObject(); + Region 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.singletonMap(new LiteralExpression("baz"), + new ValueExpression(10L))); + + handler.handleMessage(new GenericMessage("test")); + assertEquals(3, region.size()); + assertEquals(10L, region.get("baz")); } } diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml index f671560599..7c7b634932 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml @@ -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"> - + + - + - + - + + + + + - - - + + +