From 116f4934c1bc815967c50a6f30c1ed96b515f248 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 18 Feb 2013 10:22:50 +0200 Subject: [PATCH] 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 --- .../ExpressionMessageProducerSupport.java | 31 +++-- ...licationEventListeningMessageProducer.java | 115 +++++++++-------- .../src/test/java/log4j.properties | 4 +- ...ionEventListeningMessageProducerTests.java | 119 ++++++++++++++++-- .../CacheListeningMessageProducer.java | 23 ++-- .../ContinuousQueryMessageProducer.java | 23 ++-- 6 files changed, 216 insertions(+), 99 deletions(-) rename spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/SpelMessageProducerSupport.java => spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionMessageProducerSupport.java (74%) diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/SpelMessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionMessageProducerSupport.java similarity index 74% rename from spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/SpelMessageProducerSupport.java rename to spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionMessageProducerSupport.java index 0b662d7dbd..8c26c17947 100644 --- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/SpelMessageProducerSupport.java +++ b/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; } - } diff --git a/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java b/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java index 6ca0b752b3..450cad45ab 100644 --- a/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java +++ b/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 { +public class ApplicationEventListeningMessageProducer extends ExpressionMessageProducerSupport implements SmartApplicationListener { - private final Set> eventTypes = new CopyOnWriteArraySet>(); + private volatile Set> 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[] 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... eventTypes) { + Set> eventSet = new HashSet>(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 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 eventType) { + if (this.eventTypes == null) { + return true; + } + for (Class 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()); - } - } - } + diff --git a/spring-integration-event/src/test/java/log4j.properties b/spring-integration-event/src/test/java/log4j.properties index 941cbe4822..34495a6fda 100644 --- a/spring-integration-event/src/test/java/log4j.properties +++ b/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 diff --git a/spring-integration-event/src/test/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducerTests.java b/spring-integration-event/src/test/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducerTests.java index 596868c71f..972dfb8963 100644 --- a/spring-integration-event/src/test/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducerTests.java +++ b/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) 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 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 { + + private final AtomicInteger counter; + + private TestApplicationListener(AtomicInteger counter) { + this.counter = counter; + } + + public void onApplicationEvent(ApplicationEvent event) { + this.counter.incrementAndGet(); + } + } + } 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 3cbf6c97e2..4323fd805b 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 @@ -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()); } } - - + + } diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java index 5a926f21c5..5d1bf07238 100644 --- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java +++ b/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); } -} \ No newline at end of file +}