diff --git a/build.gradle b/build.gradle index 48bd58667d..2df0dfdc99 100644 --- a/build.gradle +++ b/build.gradle @@ -126,7 +126,7 @@ configure(javaprojects) { springDataMongoVersion = '1.0.0.BUILD-SNAPSHOT' springDataCommonsVersion = '1.2.0.BUILD-SNAPSHOT' springDataRedisVersion = '1.0.0.BUILD-SNAPSHOT' - springGemfireVersion = '1.0.1.RELEASE' + springGemfireVersion = '1.1.0.BUILD-SNAPSHOT' springSecurityVersion = '3.0.5.RELEASE' springWsVersion = '2.0.2.RELEASE' @@ -249,6 +249,7 @@ project('spring-integration-gemfire') { compile "org.springframework:spring-context:$springVersion" compile "org.springframework.data.gemfire:spring-gemfire:$springGemfireVersion" testCompile project(":spring-integration-stream") + testCompile project(":spring-integration-test") } repositories { mavenRepo urls: 'http://dist.gemstone.com/maven/release' // for gemfire @@ -444,6 +445,7 @@ project('spring-integration-test') { compile "junit:junit-dep:$junitVersion" compile "org.mockito:mockito-all:$mockitoVersion" compile "org.springframework:spring-context:$springVersion" + compile "org.springframework:spring-test:$springVersion" } } diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java index 8890082244..8d7296e8c0 100644 --- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java @@ -29,6 +29,7 @@ import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; +import com.gemstone.gemfire.cache.CacheClosedException; import com.gemstone.gemfire.cache.CacheListener; import com.gemstone.gemfire.cache.EntryEvent; import com.gemstone.gemfire.cache.Region; @@ -42,6 +43,7 @@ import com.gemstone.gemfire.cache.util.CacheListenerAdapter; * payloadExpression is provided, the {@link EntryEvent} itself will be the payload. * * @author Mark Fisher + * @author David Turanski * @since 2.1 */ @SuppressWarnings({"rawtypes", "unchecked"}) @@ -60,11 +62,10 @@ public class CacheListeningMessageProducer extends MessageProducerSupport { private final SpelExpressionParser parser = new SpelExpressionParser(); - public CacheListeningMessageProducer(Region region) { Assert.notNull(region, "region must not be null"); this.region = region; - this.listener = new MessageProducingCacheListener(); + this.listener = new MessageProducingCacheListener(); } @@ -95,9 +96,16 @@ public class CacheListeningMessageProducer extends MessageProducerSupport { if (logger.isInfoEnabled()) { logger.info("removing MessageProducingCacheListener from GemFire Region '" + this.region.getName() + "'"); } - this.region.getAttributesMutator().removeCacheListener(this.listener); + try { + this.region.getAttributesMutator().removeCacheListener(this.listener); + } catch (CacheClosedException e) { + if (logger.isDebugEnabled()){ + logger.debug(e.getMessage(),e); + } + } + } - + private class MessageProducingCacheListener extends CacheListenerAdapter { @@ -143,5 +151,7 @@ public class CacheListeningMessageProducer extends MessageProducerSupport { sendMessage(MessageBuilder.withPayload(payload).build()); } } + + } 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 cde54c777c..bda7c71fdd 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 @@ -16,12 +16,18 @@ package org.springframework.integration.gemfire.outbound; +import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.Map; +import java.util.Map.Entry; import org.springframework.data.gemfire.GemfireCallback; import org.springframework.data.gemfire.GemfireTemplate; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.Message; import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.util.Assert; import com.gemstone.gemfire.GemFireCheckedException; @@ -29,30 +35,37 @@ 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. + * A {@link MessageHandler} implementation that writes to a GemFire Region. The + * Message's payload must be an instance of java.util.Map. * * @author Mark Fisher + * @author David Turanski * @since 2.1 */ -public class CacheWritingMessageHandler implements MessageHandler { +public class CacheWritingMessageHandler extends AbstractMessageHandler { + private final Map cacheEntryExpressions = new LinkedHashMap(); private final GemfireTemplate gemfireTemplate = new GemfireTemplate(); - @SuppressWarnings("rawtypes") public CacheWritingMessageHandler(Region region) { Assert.notNull(region, "region must not be null"); this.gemfireTemplate.setRegion(region); - this.gemfireTemplate.afterPropertiesSet(); + this.gemfireTemplate.afterPropertiesSet(); } - - - public void handleMessage(Message message) { - // TODO: add support for more options to get key/value (SpEL?) + + @Override + public void handleMessageInternal(Message message) { Object payload = message.getPayload(); - Assert.isTrue(payload instanceof Map, "only Map payloads are supported"); - final Map map = (Map) payload; + Map cacheValues = (cacheEntryExpressions.size() > 0)?parseCacheEntries(message):null; + + if (cacheValues == null) { + Assert.isTrue(payload instanceof Map, "If cache entry expressions are not configured, then payload must be a Map"); + cacheValues = (Map) payload; + } + + final Map map = cacheValues; + this.gemfireTemplate.execute(new GemfireCallback() { @SuppressWarnings({ "rawtypes", "unchecked" }) public Object doInGemfire(Region region) throws GemFireCheckedException, GemFireException { @@ -62,4 +75,32 @@ public class CacheWritingMessageHandler implements MessageHandler { }); } + /** + * @param message + * @return + */ + private Map parseCacheEntries(Message message) { + if (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)); + } + return cacheValues; + } + } + + public void setCacheEntries(Map cacheEntries) { + + if (cacheEntryExpressions.size() > 0) { + cacheEntryExpressions.clear(); + } + + for (Entry cacheEntry : cacheEntries.entrySet()) { + this.cacheEntryExpressions.put(new SpelExpressionParser().parseExpression(cacheEntry.getKey()), + new SpelExpressionParser().parseExpression(cacheEntry.getValue())); + } + } } diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducerTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducerTests.java index a6f5039e5f..44127816d6 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducerTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducerTests.java @@ -137,5 +137,4 @@ public class CacheListeningMessageProducerTests { assertNotNull(message2); assertEquals("foo was abc", message2.getPayload()); } - } 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 074f0b5fdf..45ecb95114 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 @@ -28,11 +28,13 @@ import org.springframework.data.gemfire.RegionFactoryBean; import org.springframework.integration.Message; import org.springframework.integration.support.MessageBuilder; +import com.gemstone.bp.edu.emory.mathcs.backport.java.util.Collections; import com.gemstone.gemfire.cache.Cache; import com.gemstone.gemfire.cache.Region; /** * @author Mark Fisher + * @author David Turanski * @since 2.1 */ public class CacheWritingMessageHandlerTests { @@ -56,5 +58,31 @@ public class CacheWritingMessageHandlerTests { assertEquals(1, region.size()); assertEquals("bar", region.get("foo")); } + + @Test + public void ExpressionsWriteToCache() throws Exception { + CacheFactoryBean cacheFactoryBean = new CacheFactoryBean(); + cacheFactoryBean.afterPropertiesSet(); + Cache cache = cacheFactoryBean.getObject(); + RegionFactoryBean regionFactoryBean = new RegionFactoryBean(); + regionFactoryBean.setName("test.expressionsWriteToCache"); + regionFactoryBean.setCache(cache); + regionFactoryBean.afterPropertiesSet(); + Region region = regionFactoryBean.getObject(); + assertEquals(0, region.size()); + CacheWritingMessageHandler handler = new CacheWritingMessageHandler(region); + + Map expressions = new HashMap(); + expressions.put("'foo'", "'bar'"); + expressions.put("payload.toUpperCase()", "headers['bar'].toUpperCase()"); + handler.setCacheEntries(expressions); + + @SuppressWarnings("unchecked") + 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")); + } } diff --git a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml b/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml deleted file mode 100644 index c1ba6b7d5e..0000000000 --- a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml +++ /dev/null @@ -1,26 +0,0 @@ - - - - - - - - - - - - - - - - - diff --git a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml b/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml deleted file mode 100644 index 6f37aa49ae..0000000000 --- a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml +++ /dev/null @@ -1,28 +0,0 @@ - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/common.properties b/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/common.properties deleted file mode 100644 index 5d438b7143..0000000000 --- a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/common.properties +++ /dev/null @@ -1,5 +0,0 @@ -host=127.0.0.1 -port=55221 -region-name=people -region-query=select * from /people -correlation-header=time \ No newline at end of file diff --git a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties b/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties deleted file mode 100644 index 04392af0d1..0000000000 --- a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties +++ /dev/null @@ -1,3 +0,0 @@ -log-level=warning -name=Spring Integration GemFire World -bind-address=127.0.0.1 diff --git a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml b/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml deleted file mode 100644 index cb861d2a0f..0000000000 --- a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml +++ /dev/null @@ -1,35 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - -