Spring/spring-data-gemfire
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

SGF-726 - Impossible to define event filter for AsyncEventQueue.

Browse Source
This commit is contained in:
John Blum
2018-03-22 22:54:53 -07:00
parent 0e3ec2a124
commit fbd212467d
5 changed files with 177 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
src/main/java/org/springframework/data/gemfire/config/xml/AsyncEventQueueParser.java
View File

@@ -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))

57
src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java
View File

@@ -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<GatewayEventFilter> 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<GatewayEventFilter> 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;
}
}

2
src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.9.xsd
View File

@@ -2950,6 +2950,8 @@ use inner bean declarations.
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="event-filter" type="gatewayEventFilterType" minOccurs="0" maxOccurs="1"/>
<xsd:element name="event-substitution-filter" type="gatewayEventSubstitutionFilterType" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="name" type="xsd:string" use="optional">
<xsd:annotation>

98
src/test/java/org/springframework/data/gemfire/wan/AsyncEventQueueWithListenerIntegrationTest.java
View File

@@ -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<GatewayEventFilter> 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<Object, Object> {
private final String name;
public TestGatewayEventSubstitutionFilter(String name) {
this.name = name;
}
@Override
public Object getSubstituteValue(EntryEvent<Object, Object> event) {
return null;
}
@Override
public void close() { }
@Override
public String toString() {
return this.name;
}
}
}

20
src/test/resources/org/springframework/data/gemfire/wan/asyncEventQueueWithListener.xml
View File

@@ -1,7 +1,8 @@
<?xml version="1.0" encoding="utf-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:gfe="http://www.springframework.org/schema/gemfire"
xmlns:c="http://www.springframework.org/schema/c"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:gfe="http://www.springframework.org/schema/gemfire"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="
@@ -12,7 +13,7 @@
<util:properties id="gemfireProperties">
<prop key="name">AsyncEventQueueWithListenerTest</prop>
<prop key="log-level">config</prop>
<prop key="log-level">error</prop>
<prop key="mcast-port">0</prop>
</util:properties>
@@ -45,4 +46,19 @@
<gfe:async-event-listener ref="anotherAsyncEventListener"/>
</gfe:async-event-queue>
<bean id="testListenerOne" class="org.springframework.data.gemfire.wan.AsyncEventQueueWithListenerIntegrationTest.TestAsyncEventListener"
p:name="TestListenerOne"/>
<gfe:async-event-queue id="TestAsyncEventQueueWithFilters">
<gfe:async-event-listener ref="testListenerOne"/>
<gfe:event-filter>
<bean class="org.springframework.data.gemfire.wan.AsyncEventQueueWithListenerIntegrationTest.TestGatewayEventFilter" c:name="GatewayEventFilterOne"/>
<bean class="org.springframework.data.gemfire.wan.AsyncEventQueueWithListenerIntegrationTest.TestGatewayEventFilter" c:name="GatewayEventFilterTwo"/>
</gfe:event-filter>
<gfe:event-substitution-filter>
<bean class="org.springframework.data.gemfire.wan.AsyncEventQueueWithListenerIntegrationTest.TestGatewayEventSubstitutionFilter"
c:name="GatewayEventSubstitutionFilterOne"/>
</gfe:event-substitution-filter>
</gfe:async-event-queue>
</beans>