Merge pull request #750 from artembilan/INT-2935

* INT-2935:
  INT-2935: Improve Event Inbound Adapter

spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionMessageProducerSupport.java

@@ -1,38 +1,36 @@
 /*
- * Copyright 2002-2011 the original author or authors.
- * 
+ * Copyright 2002-2013 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;
+package org.springframework.integration.endpoint;
 
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 
 /**
+ * A {@link MessageProducerSupport} sub-class that provides {@linkplain #payloadExpression}
+ * evaluation with result as a payload for Message to send.
+ *
  * @author David Turanski
+ * @author Artem Bilan
  * @since 2.1
  *
  */
-abstract class SpelMessageProducerSupport extends MessageProducerSupport {
-	
-	private volatile Expression payloadExpression;
+public abstract class ExpressionMessageProducerSupport extends MessageProducerSupport {
 
 	private final SpelExpressionParser parser = new SpelExpressionParser();
 
-	
-	@Override
-	protected void onInit(){
-		super.onInit();
-	}
-	
+	private volatile Expression payloadExpression;
+
 	public void setPayloadExpression(String payloadExpression) {
 		if (payloadExpression == null) {
 			this.payloadExpression = null;
@@ -41,14 +39,13 @@ abstract class SpelMessageProducerSupport extends MessageProducerSupport {
 			this.payloadExpression = this.parser.parseExpression(payloadExpression);
 		}
 	}
-	
-	protected Object evaluationResult(Object payload){
+
+	protected Object evaluatePayloadExpression(Object payload){
 		Object evaluationResult = payload;
 		if (payloadExpression != null) {
 			evaluationResult = payloadExpression.getValue(payload);
 		}
 		return evaluationResult;
 	}
-		
 
 }

spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -16,85 +16,109 @@
 
 package org.springframework.integration.event.inbound;
 
+import java.util.HashSet;
 import java.util.Set;
-import java.util.concurrent.CopyOnWriteArraySet;
 
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationListener;
 import org.springframework.context.event.ApplicationContextEvent;
-import org.springframework.expression.Expression;
-import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.context.event.ApplicationEventMulticaster;
+import org.springframework.context.event.SmartApplicationListener;
+import org.springframework.context.support.AbstractApplicationContext;
+import org.springframework.core.Ordered;
 import org.springframework.integration.Message;
-import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.endpoint.ExpressionMessageProducerSupport;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 
 /**
- * An inbound Channel Adapter that passes Spring {@link ApplicationEvent ApplicationEvents} within messages.
+ * An inbound Channel Adapter that implements {@link ApplicationListener} and
+ * passes Spring {@link ApplicationEvent ApplicationEvents} within messages.
  * If a {@link #setPayloadExpression(String) payloadExpression} is provided, it will be evaluated against
  * the ApplicationEvent instance to create the Message payload. Otherwise, the event itself will be the payload.
- * 
+ *
  * @author Mark Fisher
+ * @author Artem Bilan
+ * @see ApplicationEventMulticaster
+ * @see ExpressionMessageProducerSupport
  */
-public class ApplicationEventListeningMessageProducer extends MessageProducerSupport implements ApplicationListener<ApplicationEvent> {
+public class ApplicationEventListeningMessageProducer extends ExpressionMessageProducerSupport implements SmartApplicationListener {
 
-	private final Set<Class<? extends ApplicationEvent>> eventTypes = new CopyOnWriteArraySet<Class<? extends ApplicationEvent>>();
+	private volatile Set<Class<? extends ApplicationEvent>> eventTypes;
 
-	private volatile Expression payloadExpression;
+	private ApplicationEventMulticaster applicationEventMulticaster;
 
 	private volatile boolean active;
 
-	private final SpelExpressionParser parser = new SpelExpressionParser();
-
-
 	/**
 	 * Set the list of event types (classes that extend ApplicationEvent) that
 	 * this adapter should send to the message channel. By default, all event
 	 * types will be sent.
+	 * In addition, this method re-registers the current instance as a {@link ApplicationListener}
+	 * with the {@link ApplicationEventMulticaster} which clears the listener cache. The cache will be
+	 * refreshed on the next appropriate {@link ApplicationEvent}.
+	 *
+	 * @see ApplicationEventMulticaster#addApplicationListener
+	 * @see #supportsEventType
 	 */
 	@SuppressWarnings("unchecked")
-	public void setEventTypes(Class<? extends ApplicationEvent>[] eventTypes) {
-		Assert.notEmpty(eventTypes, "at least one event type is required");
-		synchronized (this.eventTypes) {
-			this.eventTypes.clear();
-			this.eventTypes.addAll(CollectionUtils.arrayToList(eventTypes));
-		}
-	}
-
-	/**
-	 * Provide an expression to be evaluated against the received ApplicationEvent
-	 * instance (the "root object") in order to create the Message payload. If none
-	 * is provided, the ApplicationEvent itself will be used as the payload.
-	 */
-	public void setPayloadExpression(String payloadExpression) {
-		if (payloadExpression == null) {
-			this.payloadExpression = null;
-		}
-		else {
-			this.payloadExpression = this.parser.parseExpression(payloadExpression);
+	public void setEventTypes(Class<? extends ApplicationEvent>... eventTypes) {
+		Set<Class<? extends ApplicationEvent>> eventSet = new HashSet<Class<? extends ApplicationEvent>>(CollectionUtils.arrayToList(eventTypes));
+		eventSet.remove(null);
+		this.eventTypes = (eventSet.size() > 0 ? eventSet : null);
+
+		if (this.applicationEventMulticaster != null) {
+			this.applicationEventMulticaster.addApplicationListener(this);
 		}
 	}
 
+	@Override
 	public String getComponentType() {
 		return "event:inbound-channel-adapter";
 	}
 
+	@Override
+	protected void onInit() {
+		super.onInit();
+		this.applicationEventMulticaster = this.getBeanFactory()
+				.getBean(AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME, ApplicationEventMulticaster.class);
+		Assert.notNull(this.applicationEventMulticaster,
+				"To use ApplicationListeners the 'applicationEventMulticaster' bean must be supplied within ApplicationContext.");
+	}
+
 	public void onApplicationEvent(ApplicationEvent event) {
 		if (this.active || event instanceof ApplicationContextEvent) {
-			if (CollectionUtils.isEmpty(this.eventTypes)) {
-				this.sendEventAsMessage(event);
-				return;
+			if (event.getSource() instanceof Message<?>) {
+				this.sendMessage((Message<?>) event.getSource());
 			}
-			for (Class<? extends ApplicationEvent> eventType : this.eventTypes) {
-				if (eventType.isAssignableFrom(event.getClass())) {
-					this.sendEventAsMessage(event);
-					return;
-				}
+			else {
+				Object payload = this.evaluatePayloadExpression(event);
+				this.sendMessage(MessageBuilder.withPayload(payload).build());
 			}
 		}
 	}
 
+	public boolean supportsEventType(Class<? extends ApplicationEvent> eventType) {
+		if (this.eventTypes == null) {
+			return true;
+		}
+		for (Class<? extends ApplicationEvent> type : this.eventTypes) {
+			if (type.isAssignableFrom(eventType)) {
+				return true;
+			}
+		}
+		return false;
+	}
+
+	public boolean supportsSourceType(Class<?> sourceType) {
+		return true;
+	}
+
+	public int getOrder() {
+		return Ordered.LOWEST_PRECEDENCE;
+	}
+
 	@Override
 	protected void doStart() {
 		this.active = true;
@@ -105,14 +129,5 @@ public class ApplicationEventListeningMessageProducer extends MessageProducerSup
 		this.active = false;
 	}
 
-	private void sendEventAsMessage(ApplicationEvent event) {
-		if (event.getSource() instanceof Message<?>) {
-			this.sendMessage((Message<?>) event.getSource());
-		}
-		else {
-			Object payload = (this.payloadExpression != null) ? this.payloadExpression.getValue(event) : event;
-			this.sendMessage(MessageBuilder.withPayload(payload).build());
-		}
-	}
-
 }
+

spring-integration-event/src/test/java/log4j.properties

@@ -2,7 +2,7 @@ log4j.rootCategory=WARN, stdout
 
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
-log4j.appender.stdout.layout.ConversionPattern=%c{1}: %m%n
+log4j.appender.stdout.layout.ConversionPattern=%c{1}: (%t) %m%n
 
 log4j.category.org.springframework.integration=WARN
-log4j.category.org.springframework.integration.file=WARN
+log4j.category.org.springframework.integration.event=INFO

spring-integration-event/src/test/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -17,16 +17,31 @@
 package org.springframework.integration.event.inbound;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
 
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.hamcrest.Matchers;
 import org.junit.Test;
+
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.event.ApplicationEventMulticaster;
 import org.springframework.context.event.ContextClosedEvent;
 import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.context.event.ContextStartedEvent;
 import org.springframework.context.event.ContextStoppedEvent;
+import org.springframework.context.support.AbstractApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageHandlingException;
 import org.springframework.integration.channel.DirectChannel;
@@ -35,10 +50,12 @@ import org.springframework.integration.core.PollableChannel;
 import org.springframework.integration.event.core.MessagingEvent;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.test.util.TestUtils;
 
 /**
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class ApplicationEventListeningMessageProducerTests {
 
@@ -50,7 +67,9 @@ public class ApplicationEventListeningMessageProducerTests {
 		adapter.start();
 		Message<?> message1 = channel.receive(0);
 		assertNull(message1);
+		assertTrue(adapter.supportsEventType(TestApplicationEvent1.class));
 		adapter.onApplicationEvent(new TestApplicationEvent1());
+		assertTrue(adapter.supportsEventType(TestApplicationEvent2.class));
 		adapter.onApplicationEvent(new TestApplicationEvent2());
 		Message<?> message2 = channel.receive(20);
 		assertNotNull(message2);
@@ -66,17 +85,29 @@ public class ApplicationEventListeningMessageProducerTests {
 		QueueChannel channel = new QueueChannel();
 		ApplicationEventListeningMessageProducer adapter = new ApplicationEventListeningMessageProducer();
 		adapter.setOutputChannel(channel);
-		adapter.setEventTypes(new Class[]{TestApplicationEvent1.class});
+		adapter.setEventTypes(TestApplicationEvent1.class);
 		adapter.start();
 		Message<?> message1 = channel.receive(0);
 		assertNull(message1);
+		assertTrue(adapter.supportsEventType(TestApplicationEvent1.class));
 		adapter.onApplicationEvent(new TestApplicationEvent1());
-		adapter.onApplicationEvent(new TestApplicationEvent2());
+		assertFalse(adapter.supportsEventType(TestApplicationEvent2.class));
 		Message<?> message2 = channel.receive(20);
 		assertNotNull(message2);
 		assertEquals("event1", ((ApplicationEvent) message2.getPayload()).getSource());
-		Message<?> message3 = channel.receive(0);
-		assertNull(message3);
+		assertNull(channel.receive(0));
+
+		adapter.setEventTypes((Class<? extends ApplicationEvent>) null);
+		assertTrue(adapter.supportsEventType(TestApplicationEvent1.class));
+		assertTrue(adapter.supportsEventType(TestApplicationEvent2.class));
+
+		adapter.setEventTypes(null, TestApplicationEvent2.class, null);
+		assertFalse(adapter.supportsEventType(TestApplicationEvent1.class));
+		assertTrue(adapter.supportsEventType(TestApplicationEvent2.class));
+
+		adapter.setEventTypes(null, null);
+		assertTrue(adapter.supportsEventType(TestApplicationEvent1.class));
+		assertTrue(adapter.supportsEventType(TestApplicationEvent2.class));
 	}
 
 	@Test
@@ -148,7 +179,7 @@ public class ApplicationEventListeningMessageProducerTests {
 		assertEquals("test", message2.getPayload());
 	}
 
-	@Test(expected=MessageHandlingException.class)
+	@Test(expected = MessageHandlingException.class)
 	public void anyApplicationEventCausesExceptionWithErrorHandling() {
 		DirectChannel channel = new DirectChannel();
 		channel.subscribe(new AbstractReplyProducingMessageHandler() {
@@ -170,6 +201,67 @@ public class ApplicationEventListeningMessageProducerTests {
 		adapter.onApplicationEvent(new TestApplicationEvent1());
 	}
 
+	@Test
+	@SuppressWarnings({"unchecked", "serial"})
+	public void testInt2935CheckRetrieverCache() {
+		GenericApplicationContext ctx = TestUtils.createTestApplicationContext();
+		ConfigurableListableBeanFactory beanFactory = ctx.getBeanFactory();
+
+		QueueChannel channel = new QueueChannel();
+		ApplicationEventListeningMessageProducer listenerMessageProducer = new ApplicationEventListeningMessageProducer();
+		listenerMessageProducer.setOutputChannel(channel);
+		listenerMessageProducer.setEventTypes(TestApplicationEvent2.class);
+		beanFactory.registerSingleton("testListenerMessageProducer", listenerMessageProducer);
+
+		AtomicInteger listenerCounter = new AtomicInteger();
+		beanFactory.registerSingleton("testListener", new TestApplicationListener(listenerCounter));
+
+		ctx.refresh();
+
+		ApplicationEventMulticaster multicaster =
+				ctx.getBean(AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME, ApplicationEventMulticaster.class);
+		Map<?, ?> retrieverCache = TestUtils.getPropertyValue(multicaster, "retrieverCache", Map.class);
+
+		ctx.publishEvent(new TestApplicationEvent1());
+
+		/*
+		 *  Previously, the retrieverCache grew unnecessarily; the adapter was added to the cache for each event type,
+		 *  event if not supported.
+		 */
+		assertEquals(2, retrieverCache.size());
+		for (Object key : retrieverCache.keySet()) {
+			Class<? extends ApplicationEvent> event = TestUtils.getPropertyValue(key, "eventType", Class.class);
+			assertThat(event, Matchers.is(Matchers.isOneOf(ContextRefreshedEvent.class, TestApplicationEvent1.class)));
+			Set<?> listeners = TestUtils.getPropertyValue(retrieverCache.get(key), "applicationListenerBeans", Set.class);
+			assertEquals(1, listeners.size());
+			assertEquals("testListener", listeners.iterator().next());
+		}
+
+		TestApplicationEvent2 event2 = new TestApplicationEvent2();
+		ctx.publishEvent(event2);
+		assertEquals(3, retrieverCache.size());
+		for (Object key : retrieverCache.keySet()) {
+			Class<?> event = TestUtils.getPropertyValue(key, "eventType", Class.class);
+			if (TestApplicationEvent2.class.isAssignableFrom(event)) {
+				Set<?> listeners = TestUtils.getPropertyValue(retrieverCache.get(key), "applicationListenerBeans", Set.class);
+				assertEquals(2, listeners.size());
+				for (Object listener : listeners) {
+					assertThat((String) listener, Matchers.is(Matchers.isOneOf("testListenerMessageProducer", "testListener")));
+				}
+				break;
+			}
+		}
+
+		ctx.publishEvent(new ApplicationEvent("Some event") {});
+
+		assertEquals(4, listenerCounter.get());
+
+		final Message<?> receive = channel.receive(10);
+		assertNotNull(receive);
+		assertSame(event2, receive.getPayload());
+		assertNull(channel.receive(1));
+	}
+
 
 	@SuppressWarnings("serial")
 	private static class TestApplicationEvent1 extends ApplicationEvent {
@@ -179,7 +271,6 @@ public class ApplicationEventListeningMessageProducerTests {
 		}
 	}
 
-
 	@SuppressWarnings("serial")
 	private static class TestApplicationEvent2 extends ApplicationEvent {
 
@@ -188,7 +279,6 @@ public class ApplicationEventListeningMessageProducerTests {
 		}
 	}
 
-
 	@SuppressWarnings("serial")
 	private static class TestMessagingEvent extends ApplicationEvent {
 
@@ -197,4 +287,17 @@ public class ApplicationEventListeningMessageProducerTests {
 		}
 	}
 
+	private static class TestApplicationListener implements ApplicationListener<ApplicationEvent> {
+
+		private final AtomicInteger counter;
+
+		private TestApplicationListener(AtomicInteger counter) {
+			this.counter = counter;
+		}
+
+		public void onApplicationEvent(ApplicationEvent event) {
+			this.counter.incrementAndGet();
+		}
+	}
+
 }

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