INT-2049 added namespace support, also add SpEL support to CacheWritingMessageHandler and general cleanup

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"
     }
 }
 

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());
 		}
 	}
+	
+	
 
 }

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<Expression, Expression> cacheEntryExpressions = new LinkedHashMap<Expression, Expression>();
 
 	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<Object>() {
 			@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<Object, Object> parseCacheEntries(Message<?> message) {
+		if (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));
+			}
+			return cacheValues;
+		}
+	}
+
+	public void setCacheEntries(Map<String, String> cacheEntries) {
+		
+		if (cacheEntryExpressions.size() > 0) {
+			cacheEntryExpressions.clear();
+		}
+		
+		for (Entry<String, String> cacheEntry : cacheEntries.entrySet()) {
+			this.cacheEntryExpressions.put(new SpelExpressionParser().parseExpression(cacheEntry.getKey()),
+					new SpelExpressionParser().parseExpression(cacheEntry.getValue()));
+		}
+	}
 }

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());		
 	}
-
 }

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<String, String> regionFactoryBean = new RegionFactoryBean<String, String>();
+		regionFactoryBean.setName("test.expressionsWriteToCache");
+		regionFactoryBean.setCache(cache);
+		regionFactoryBean.afterPropertiesSet();
+		Region<String, String> region = regionFactoryBean.getObject();
+		assertEquals(0, region.size());
+		CacheWritingMessageHandler handler = new CacheWritingMessageHandler(region);
+		
+		Map<String, String> expressions = new HashMap<String, String>();
+		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"));
+	}
 
 }

spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml

@@ -1,26 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<beans xmlns="http://www.springframework.org/schema/beans"
-       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-       xmlns:p="http://www.springframework.org/schema/p"
-       xmlns:util="http://www.springframework.org/schema/util"
-       xmlns:int="http://www.springframework.org/schema/integration"
-       xmlns:gfe="http://www.springframework.org/schema/gemfire" xmlns:context="http://www.springframework.org/schema/context"
-       xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
-	    http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
-		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
-		http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
-
-
-    <context:component-scan base-package="org.springframework.integration.gemfire.inbound.cq.client"/>
-
-    <context:property-placeholder location="org/springframework/integration/gemfire/inbound/cq/common.properties"/>
-
-    <!--<util:properties id="props" location="org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties"/>-->
-
-
-     
-    <int:channel id="cqIn"/>
-
-    <int:service-activator input-channel="cqIn" ref="cqServiceActivator"/>
-
-</beans>

spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml

@@ -1,28 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<beans xmlns="http://www.springframework.org/schema/beans"
-       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-       xmlns:p="http://www.springframework.org/schema/p"
-       xmlns:util="http://www.springframework.org/schema/util"
-       xmlns:int="http://www.springframework.org/schema/integration"
-       xmlns:gfe="http://www.springframework.org/schema/gemfire" xmlns:context="http://www.springframework.org/schema/context"
-       xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
-	    http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
-		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
-		http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
-
-
-    <!-- infrastrcture-->
-    <context:component-scan base-package="org.springframework.integration.gemfire.inbound.cq.server"/>
-    <context:property-placeholder location="org/springframework/integration/gemfire/inbound/cq/common.properties"/>
-
-    <!-- setup the cache-->
-    <util:properties id="props" location="org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties"/>
-    <gfe:cache properties-ref="props" id="c"/>
-
-    <!-- tx manager-->
-    <gfe:transaction-manager cache-ref="c"/>
-
-    <!-- region -->
-    <gfe:replicated-region id="r" name="${region-name}" cache-ref="c" />
-
-</beans>

spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/common.properties

@@ -1,5 +0,0 @@
-host=127.0.0.1
-port=55221
-region-name=people
-region-query=select * from /people
-correlation-header=time

spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties

@@ -1,3 +0,0 @@
-log-level=warning
-name=Spring Integration GemFire World
-bind-address=127.0.0.1

spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml

@@ -1,35 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<beans xmlns="http://www.springframework.org/schema/beans"
-       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-       xmlns:p="http://www.springframework.org/schema/p"
-       xmlns:util="http://www.springframework.org/schema/util"
-       xmlns:int="http://www.springframework.org/schema/integration"
-       xmlns:gfe="http://www.springframework.org/schema/gemfire" xmlns:context="http://www.springframework.org/schema/context"
-       xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
-	    http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
-		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
-		http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
-
-    <context:property-placeholder location="org/springframework/integration/gemfire/inbound/cq/common.properties"/>
-    
-    <context:component-scan base-package="org.springframework.integration.gemfire.store.messagegroupstore"/>
-
-    <int:channel id="i"/>
-
-    <int:aggregator  release-strategy="releaseStrategy" correlation-strategy="correlationStrategy" message-store="gemfireMessageGroupStore" input-channel="i" output-channel="o"  />
-
-    <int:channel id="o"/>
-
-    <int:service-activator input-channel="o" ref="messageGroupStoreActivator" />
-
-
-    <util:properties id="props" location="org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties"/>
-
-	<gfe:cache properties-ref="props" id="c"/>
-    <gfe:transaction-manager cache-ref="c"/>
-
-    <gfe:replicated-region id="unmarkedRegion"      cache-ref="c"/>
-    <gfe:replicated-region id="markedRegion"        cache-ref="c"/>
-    <gfe:replicated-region id="messageGroupRegion"  cache-ref="c"/>
-
-</beans>