INT-2935: Improve Event Inbound Adapter

Previously, the `ApplicationEventListeningMessageProducer` accepted
all `ApplicationEvent`'s and than filtered them.
This caused some `ApplicationEventMulticaster.retrieverCache` overhead.

* Improve `ApplicationEventListeningMessageProducer` to `implements SmartApplicationListener`.

    This allows filtering earlier on first the appropriate `ApplicationEvent`
    using `ApplicationEventListeningMessageProducer#supportsEventType`
    and caching the `ApplicationListener` only for that `ApplicationEvent`.

* Re-register `ApplicationEventListeningMessageProducer` in the `ApplicationEventMulticaster`
    when  `ApplicationEventListeningMessageProducer#setEventTypes` is invoked
    to clear the `ApplicationEventMulticaster.retrieverCache`.

* Move `org.springframework.integration.gemfire.inbound.SpelMessageProducerSupport` to core `ExpressionMessageProducerSupport`.

* Add test for the new logic int `ApplicationEventListeningMessageProducer` and its behavior with respect to the `ApplicationEventMulticaster.retrieverCache`.

JIRA: https://jira.springsource.org/browse/INT-2935

INT-2935: EMPS JavaDoc & AELMP ReadWriteLock

INT-2935: AELMP#eventTypes changing 'barrier'

INT-2935: avoid 'eventTypes' mutation afterwards

INT-2935: Polishing according PR comments

INT-2935: Fix NPE in the `setEventTypes`

Minor Polishing

   Polish java docs/comments
   Remove compiler warnings

spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java

@@ -22,6 +22,7 @@ import java.util.Set;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.springframework.integration.endpoint.ExpressionMessageProducerSupport;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.util.Assert;
 
@@ -37,13 +38,13 @@ import com.gemstone.gemfire.cache.util.CacheListenerAdapter;
  * enum for all options. A SpEL expression may be provided to generate a Message payload by
  * evaluating that expression against the {@link EntryEvent} instance as the root object. If no
  * payloadExpression is provided, the {@link EntryEvent} itself will be the payload.
- * 
+ *
  * @author Mark Fisher
  * @author David Turanski
  * @since 2.1
  */
 @SuppressWarnings({"rawtypes", "unchecked"})
-public class CacheListeningMessageProducer extends SpelMessageProducerSupport {
+public class CacheListeningMessageProducer extends ExpressionMessageProducerSupport {
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
@@ -58,7 +59,7 @@ public class CacheListeningMessageProducer extends SpelMessageProducerSupport {
 	public CacheListeningMessageProducer(Region<?, ?> region) {
 		Assert.notNull(region, "region must not be null");
 		this.region = region;
-		this.listener = new MessageProducingCacheListener();   
+		this.listener = new MessageProducingCacheListener();
 	}
 
 
@@ -81,16 +82,16 @@ public class CacheListeningMessageProducer extends SpelMessageProducerSupport {
 		if (logger.isInfoEnabled()) {
 			logger.info("removing MessageProducingCacheListener from GemFire Region '" + this.region.getName() + "'");
 		}
-		try { 
+		try {
 			this.region.getAttributesMutator().removeCacheListener(this.listener);
 		} catch (CacheClosedException e) {
 			if (logger.isDebugEnabled()){
 				logger.debug(e.getMessage(),e);
 			}
 		}
-		 
+
 	}
- 
+
 	private class MessageProducingCacheListener extends CacheListenerAdapter {
 
 		@Override
@@ -121,16 +122,16 @@ public class CacheListeningMessageProducer extends SpelMessageProducerSupport {
 			}
 		}
 
-		private void processEvent(EntryEvent event) { 
-				this.publish(evaluationResult(event));
-			 
+		private void processEvent(EntryEvent event) {
+				this.publish(evaluatePayloadExpression(event));
+
 		}
 
 		private void publish(Object payload) {
 			sendMessage(MessageBuilder.withPayload(payload).build());
 		}
 	}
-	
-	
+
+
 
 }

spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java

@@ -26,6 +26,7 @@ import org.springframework.data.gemfire.listener.ContinuousQueryDefinition;
 import org.springframework.data.gemfire.listener.ContinuousQueryListener;
 import org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer;
 import org.springframework.integration.Message;
+import org.springframework.integration.endpoint.ExpressionMessageProducerSupport;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.util.Assert;
 
@@ -36,13 +37,13 @@ import com.gemstone.gemfire.cache.query.CqEvent;
  * constantly evaluated against a cache
  * {@link com.gemstone.gemfire.cache.Region}. This is much faster than
  * re-querying the cache manually.
- * 
+ *
  * @author Josh Long
  * @author David Turanski
  * @since 2.1
- * 
+ *
  */
-public class ContinuousQueryMessageProducer extends SpelMessageProducerSupport implements ContinuousQueryListener {
+public class ContinuousQueryMessageProducer extends ExpressionMessageProducerSupport implements ContinuousQueryListener {
 	private static Log logger = LogFactory.getLog(ContinuousQueryMessageProducer.class);
 
 	private final String query;
@@ -57,7 +58,7 @@ public class ContinuousQueryMessageProducer extends SpelMessageProducerSupport i
 			CqEventType.UPDATED));
 
 	/**
-	 * 
+	 *
 	 * @param queryListenerContainer a {@link org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer}
 	 * @param query the query string
 	 */
@@ -69,7 +70,7 @@ public class ContinuousQueryMessageProducer extends SpelMessageProducerSupport i
 	}
 
 	/**
-	 * 
+	 *
 	 * @param queryName optional query name
 	 */
 	public void setQueryName(String queryName) {
@@ -77,7 +78,7 @@ public class ContinuousQueryMessageProducer extends SpelMessageProducerSupport i
 	}
 
 	/**
-	 * 
+	 *
 	 * @param durable true if the query is a durable subscription
 	 */
 	public void setDurable(boolean durable) {
@@ -102,7 +103,7 @@ public class ContinuousQueryMessageProducer extends SpelMessageProducerSupport i
 
 	/*
 	 * (non-Javadoc)
-	 * 
+	 *
 	 * @see
 	 * org.springframework.data.gemfire.listener.QueryListener#onEvent(com.gemstone
 	 * .gemfire.cache.query.CqEvent)
@@ -113,17 +114,17 @@ public class ContinuousQueryMessageProducer extends SpelMessageProducerSupport i
 				logger.debug(String.format("processing cq event key [%s] event [%s]", event.getQueryOperation()
 						.toString(), event.getKey()));
 			}
-			Message<?> cqEventMessage = MessageBuilder.withPayload(evaluationResult(event)).build();
+			Message<?> cqEventMessage = MessageBuilder.withPayload(evaluatePayloadExpression(event)).build();
 			sendMessage(cqEventMessage);
 		}
 	}
 
 	private boolean isEventSupported(CqEvent event) {
-		 
-		 String eventName = event.getQueryOperation().toString() + 
+
+		 String eventName = event.getQueryOperation().toString() +
 		 	(event.getQueryOperation().toString().endsWith("Y")? "ED" : "D");
 		 CqEventType eventType = CqEventType.valueOf(eventName);
 		 return supportedEventTypes.contains(eventType);
 	}
 
-}
+}

spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/SpelMessageProducerSupport.java

@@ -1,54 +0,0 @@
-/*
- * Copyright 2002-2011 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. You may obtain a copy of the License at
- * 
- * http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
- * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations under the License.
- */
-package org.springframework.integration.gemfire.inbound;
-
-import org.springframework.expression.Expression;
-import org.springframework.expression.spel.standard.SpelExpressionParser;
-import org.springframework.integration.endpoint.MessageProducerSupport;
-
-/**
- * @author David Turanski
- * @since 2.1
- *
- */
-abstract class SpelMessageProducerSupport extends MessageProducerSupport {
-	
-	private volatile Expression payloadExpression;
-
-	private final SpelExpressionParser parser = new SpelExpressionParser();
-
-	
-	@Override
-	protected void onInit(){
-		super.onInit();
-	}
-	
-	public void setPayloadExpression(String payloadExpression) {
-		if (payloadExpression == null) {
-			this.payloadExpression = null;
-		}
-		else {
-			this.payloadExpression = this.parser.parseExpression(payloadExpression);
-		}
-	}
-	
-	protected Object evaluationResult(Object payload){
-		Object evaluationResult = payload;
-		if (payloadExpression != null) {
-			evaluationResult = payloadExpression.getValue(payload);
-		}
-		return evaluationResult;
-	}
-		
-
-}