From fbd212467d08cedc81183c703ad8e1c4dea9e01d Mon Sep 17 00:00:00 2001 From: John Blum Date: Thu, 22 Mar 2018 22:54:53 -0700 Subject: [PATCH] SGF-726 - Impossible to define event filter for AsyncEventQueue. --- .../config/xml/AsyncEventQueueParser.java | 29 ++++-- .../wan/AsyncEventQueueFactoryBean.java | 57 ++++++++--- .../gemfire/config/spring-gemfire-1.9.xsd | 2 + ...EventQueueWithListenerIntegrationTest.java | 98 +++++++++++++++++-- .../wan/asyncEventQueueWithListener.xml | 20 +++- 5 files changed, 177 insertions(+), 29 deletions(-) diff --git a/src/main/java/org/springframework/data/gemfire/config/xml/AsyncEventQueueParser.java b/src/main/java/org/springframework/data/gemfire/config/xml/AsyncEventQueueParser.java index 80bc5576..667bb09e 100644 --- a/src/main/java/org/springframework/data/gemfire/config/xml/AsyncEventQueueParser.java +++ b/src/main/java/org/springframework/data/gemfire/config/xml/AsyncEventQueueParser.java @@ -20,7 +20,6 @@ import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; -import org.springframework.data.gemfire.GemfireUtils; import org.springframework.data.gemfire.util.SpringUtils; import org.springframework.data.gemfire.wan.AsyncEventQueueFactoryBean; import org.springframework.util.StringUtils; @@ -50,30 +49,44 @@ class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser { */ @Override protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + builder.setLazyInit(false); parseAsyncEventListener(element, parserContext, builder); parseCache(element, builder); parseDiskStore(element, builder); + ParsingUtils.setPropertyValue(element, builder, "enable-batch-conflation", "batchConflationEnabled"); + ParsingUtils.setPropertyValue(element, builder, "batch-conflation-enabled"); ParsingUtils.setPropertyValue(element, builder, "batch-size"); + ParsingUtils.setPropertyValue(element, builder, "batch-time-interval"); + ParsingUtils.setPropertyValue(element, builder, "disk-synchronous"); + ParsingUtils.setPropertyValue(element, builder, "dispatcher-threads"); ParsingUtils.setPropertyValue(element, builder, "maximum-queue-memory"); + ParsingUtils.setPropertyValue(element, builder, "order-policy"); ParsingUtils.setPropertyValue(element, builder, "parallel"); ParsingUtils.setPropertyValue(element, builder, "persistent"); - if (GemfireUtils.GEMFIRE_VERSION.compareTo("7.0.1") >= 0) { - ParsingUtils.setPropertyValue(element, builder, "enable-batch-conflation", "batchConflationEnabled"); - ParsingUtils.setPropertyValue(element, builder, "batch-conflation-enabled"); - ParsingUtils.setPropertyValue(element, builder, "batch-time-interval"); - ParsingUtils.setPropertyValue(element, builder, "disk-synchronous"); - ParsingUtils.setPropertyValue(element, builder, "dispatcher-threads"); - ParsingUtils.setPropertyValue(element, builder, "order-policy"); + Element eventFilterElement = DomUtils.getChildElementByTagName(element, "event-filter"); + + if (eventFilterElement != null) { + builder.addPropertyValue("gatewayEventFilters", + ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, eventFilterElement, builder)); + } + + Element eventSubstitutionFilterElement = + DomUtils.getChildElementByTagName(element, "event-substitution-filter"); + + if (eventSubstitutionFilterElement != null) { + builder.addPropertyValue("gatewayEventSubstitutionFilter", + ParsingUtils.parseRefOrSingleNestedBeanDeclaration(parserContext, eventSubstitutionFilterElement, builder)); } ParsingUtils.setPropertyValue(element, builder, NAME_ATTRIBUTE); if (!StringUtils.hasText(element.getAttribute(NAME_ATTRIBUTE))) { if (element.getParentNode().getNodeName().endsWith("region")) { + Element region = (Element) element.getParentNode(); String regionName = StringUtils.hasText(region.getAttribute(NAME_ATTRIBUTE)) diff --git a/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java b/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java index a848dbb6..651a596b 100644 --- a/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java @@ -13,9 +13,12 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.data.gemfire.wan; -import org.springframework.util.Assert; +import static org.springframework.data.gemfire.util.CollectionUtils.nullSafeList; + +import java.util.List; import com.gemstone.gemfire.cache.Cache; import com.gemstone.gemfire.cache.CacheClosedException; @@ -23,10 +26,14 @@ import com.gemstone.gemfire.cache.asyncqueue.AsyncEventListener; import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueue; import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueueFactory; import com.gemstone.gemfire.cache.util.Gateway; +import com.gemstone.gemfire.cache.wan.GatewayEventFilter; +import com.gemstone.gemfire.cache.wan.GatewayEventSubstitutionFilter; + +import org.springframework.util.Assert; /** * FactoryBean for creating GemFire {@link AsyncEventQueue}s. - * + * * @author David Turanski * @author John Blum */ @@ -42,51 +49,56 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean< private Boolean parallel; private Boolean persistent; + private GatewayEventSubstitutionFilter gatewayEventSubstitutionFilter; + private Integer batchSize; private Integer batchTimeInterval; private Integer dispatcherThreads; private Integer maximumQueueMemory; + private List gatewayEventFilters; + private String diskStoreReference; private String orderPolicy; /** * Constructs an instance of the AsyncEventQueueFactoryBean for creating an GemFire AsyncEventQueue. - * + * * @param cache the GemFire Cache reference. * @see #AsyncEventQueueFactoryBean(com.gemstone.gemfire.cache.Cache, com.gemstone.gemfire.cache.asyncqueue.AsyncEventListener) */ - public AsyncEventQueueFactoryBean(final Cache cache) { + public AsyncEventQueueFactoryBean(Cache cache) { this(cache, null); } /** * Constructs an instance of the AsyncEventQueueFactoryBean for creating an GemFire AsyncEventQueue. - * + * * @param cache the GemFire Cache reference. * @param asyncEventListener required {@link AsyncEventListener} */ - public AsyncEventQueueFactoryBean(final Cache cache, final AsyncEventListener asyncEventListener) { + public AsyncEventQueueFactoryBean(Cache cache, AsyncEventListener asyncEventListener) { super(cache); setAsyncEventListener(asyncEventListener); } @Override public AsyncEventQueue getObject() throws Exception { - return asyncEventQueue; + return this.asyncEventQueue; } @Override public Class getObjectType() { - return AsyncEventQueue.class; + return this.asyncEventQueue != null ? this.asyncEventQueue.getClass() : AsyncEventQueue.class; } @Override protected void doInit() { + Assert.notNull(this.asyncEventListener, "The AsyncEventListener cannot be null."); - AsyncEventQueueFactory asyncEventQueueFactory = (this.factory != null ? (AsyncEventQueueFactory) factory - : cache.createAsyncEventQueueFactory()); + AsyncEventQueueFactory asyncEventQueueFactory = + this.factory != null ? (AsyncEventQueueFactory) this.factory : this.cache.createAsyncEventQueueFactory(); if (batchSize != null) { asyncEventQueueFactory.setBatchSize(batchSize); @@ -112,6 +124,14 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean< asyncEventQueueFactory.setDiskSynchronous(diskSynchronous); } + for (GatewayEventFilter gatewayEventFilter : nullSafeList(gatewayEventFilters)) { + asyncEventQueueFactory.addGatewayEventFilter(gatewayEventFilter); + } + + if (gatewayEventSubstitutionFilter != null) { + asyncEventQueueFactory.setGatewayEventSubstitutionListener(gatewayEventSubstitutionFilter); + } + if (maximumQueueMemory != null) { asyncEventQueueFactory.setMaximumQueueMemory(maximumQueueMemory); } @@ -119,6 +139,7 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean< asyncEventQueueFactory.setParallel(isParallelEventQueue()); if (orderPolicy != null) { + Assert.isTrue(isSerialEventQueue(), "Order Policy cannot be used with a Parallel Event Queue."); Assert.isTrue(VALID_ORDER_POLICIES.contains(orderPolicy.toUpperCase()), String.format( @@ -131,12 +152,13 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean< asyncEventQueueFactory.setPersistent(persistent); } - asyncEventQueue = asyncEventQueueFactory.create(getName(), this.asyncEventListener); + this.asyncEventQueue = asyncEventQueueFactory.create(getName(), this.asyncEventListener); } @Override public void destroy() throws Exception { - if (!cache.isClosed()) { + + if (!this.cache.isClosed()) { try { this.asyncEventListener.close(); } @@ -146,8 +168,10 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean< } public final void setAsyncEventListener(AsyncEventListener listener) { + Assert.state(this.asyncEventQueue == null, "Setting an AsyncEventListener is not allowed once the AsyncEventQueue has been created."); + this.asyncEventListener = listener; } @@ -204,6 +228,14 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean< this.dispatcherThreads = dispatcherThreads; } + public void setGatewayEventFilters(List gatewayEventFilters) { + this.gatewayEventFilters = gatewayEventFilters; + } + + public void setGatewayEventSubstitutionFilter(GatewayEventSubstitutionFilter gatewayEventSubstitutionFilter) { + this.gatewayEventSubstitutionFilter = gatewayEventSubstitutionFilter; + } + public void setMaximumQueueMemory(Integer maximumQueueMemory) { this.maximumQueueMemory = maximumQueueMemory; } @@ -234,5 +266,4 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean< public void setPersistent(Boolean persistent) { this.persistent = persistent; } - } diff --git a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.9.xsd b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.9.xsd index 67c24bfe..e360a7fb 100644 --- a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.9.xsd +++ b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.9.xsd @@ -2950,6 +2950,8 @@ use inner bean declarations. + + diff --git a/src/test/java/org/springframework/data/gemfire/wan/AsyncEventQueueWithListenerIntegrationTest.java b/src/test/java/org/springframework/data/gemfire/wan/AsyncEventQueueWithListenerIntegrationTest.java index ba39c75d..81d9f524 100644 --- a/src/test/java/org/springframework/data/gemfire/wan/AsyncEventQueueWithListenerIntegrationTest.java +++ b/src/test/java/org/springframework/data/gemfire/wan/AsyncEventQueueWithListenerIntegrationTest.java @@ -16,6 +16,7 @@ package org.springframework.data.gemfire.wan; +import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; @@ -23,8 +24,18 @@ import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import java.util.List; + import javax.annotation.Resource; +import com.gemstone.gemfire.cache.EntryEvent; +import com.gemstone.gemfire.cache.asyncqueue.AsyncEvent; +import com.gemstone.gemfire.cache.asyncqueue.AsyncEventListener; +import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueue; +import com.gemstone.gemfire.cache.wan.GatewayEventFilter; +import com.gemstone.gemfire.cache.wan.GatewayEventSubstitutionFilter; +import com.gemstone.gemfire.cache.wan.GatewayQueueEvent; +import com.gemstone.gemfire.cache.wan.GatewaySender; + import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.data.gemfire.test.GemfireTestApplicationContextInitializer; @@ -33,11 +44,6 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.Assert; import org.springframework.util.StringUtils; -import com.gemstone.gemfire.cache.asyncqueue.AsyncEvent; -import com.gemstone.gemfire.cache.asyncqueue.AsyncEventListener; -import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueue; -import com.gemstone.gemfire.cache.wan.GatewaySender; - /** * The AsyncEventQueueWithListenerIntegrationTest class is a test suite of test cases testing the circular references * between an Async Event Queue and a registered AsyncEventListener that refers back to the Async Event Queue @@ -65,8 +71,12 @@ public class AsyncEventQueueWithListenerIntegrationTest { @Resource(name = "Q3") private AsyncEventQueue queueThree; + @Resource(name = "TestAsyncEventQueueWithFilters") + private AsyncEventQueue queueWithFilters; + @Test public void testAsyncEventQueueOneAndListenerConfiguration() { + assertNotNull(queueOne); assertEquals("QueueOne", queueOne.getId()); assertFalse(queueOne.isPersistent()); @@ -78,6 +88,7 @@ public class AsyncEventQueueWithListenerIntegrationTest { } @Test public void testAsyncEventQueueTwoAndListenerConfiguration() { + assertNotNull(queueTwo); assertEquals("QueueTwo", queueTwo.getId()); assertFalse(queueTwo.isPersistent()); @@ -90,6 +101,7 @@ public class AsyncEventQueueWithListenerIntegrationTest { @Test public void testAsyncEventQueueThreeAndListenerConfiguration() { + assertNotNull(queueThree); assertEquals("QueueThree", queueThree.getId()); assertFalse(queueThree.isPersistent()); @@ -100,6 +112,31 @@ public class AsyncEventQueueWithListenerIntegrationTest { assertSame(queueThree, ((TestAsyncEventListener) queueThree.getAsyncEventListener()).getQueue()); } + @Test + public void asyncEventQueueWithFiltersIsConfiguredProperly() { + + assertThat(queueWithFilters).isNotNull(); + assertThat(queueWithFilters.getId()).isEqualTo("TestAsyncEventQueueWithFilters"); + + AsyncEventListener listener = queueWithFilters.getAsyncEventListener(); + + assertThat(listener).isNotNull(); + assertThat(listener.toString()).isEqualTo("TestListenerOne"); + + List gatewayEventFilters = queueWithFilters.getGatewayEventFilters(); + + assertThat(gatewayEventFilters).isNotNull(); + assertThat(gatewayEventFilters).hasSize(2); + assertThat(gatewayEventFilters.get(0).toString()).isEqualTo("GatewayEventFilterOne"); + assertThat(gatewayEventFilters.get(1).toString()).isEqualTo("GatewayEventFilterTwo"); + + GatewayEventSubstitutionFilter gatewayEventSubstitutionFilter = + queueWithFilters.getGatewayEventSubstitutionFilter(); + + assertThat(gatewayEventSubstitutionFilter).isNotNull(); + assertThat(gatewayEventSubstitutionFilter.toString()).isEqualTo("GatewayEventSubstitutionFilterOne"); + } + /** * The QueueAsyncEventListener class is an implementation of the AsyncEventListener interface that contains * a reference to the AsyncEventQueue upon which it is registered. @@ -159,7 +196,56 @@ public class AsyncEventQueueWithListenerIntegrationTest { public String toString() { return (StringUtils.hasText(getName()) ? getName() : getClass().getName()); } - } + public static class TestGatewayEventFilter implements GatewayEventFilter { + + private final String name; + + public TestGatewayEventFilter(String name) { + this.name = name; + } + + @Override + public boolean beforeEnqueue(GatewayQueueEvent event) { + return false; + } + + @Override + public boolean beforeTransmit(GatewayQueueEvent event) { + return false; + } + + @Override + public void afterAcknowledgement(GatewayQueueEvent event) { } + + @Override + public void close() { } + + @Override + public String toString() { + return this.name; + } + } + + public static class TestGatewayEventSubstitutionFilter implements GatewayEventSubstitutionFilter { + + private final String name; + + public TestGatewayEventSubstitutionFilter(String name) { + this.name = name; + } + @Override + public Object getSubstituteValue(EntryEvent event) { + return null; + } + + @Override + public void close() { } + + @Override + public String toString() { + return this.name; + } + } } diff --git a/src/test/resources/org/springframework/data/gemfire/wan/asyncEventQueueWithListener.xml b/src/test/resources/org/springframework/data/gemfire/wan/asyncEventQueueWithListener.xml index 7bc8032e..b9b2991a 100644 --- a/src/test/resources/org/springframework/data/gemfire/wan/asyncEventQueueWithListener.xml +++ b/src/test/resources/org/springframework/data/gemfire/wan/asyncEventQueueWithListener.xml @@ -1,7 +1,8 @@ AsyncEventQueueWithListenerTest - config + error 0 @@ -45,4 +46,19 @@ + + + + + + + + + + + + +