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

SGF-500 - Add support for the 'ignoreEvictionAndExpiration' configuration property on AsyncEventQueue and AsyncEventQueueFactory

Browse Source
This commit is contained in:
John Blum
2016-05-18 19:13:10 -07:00
parent 3f84baa378
commit 2fe1029abf
6 changed files with 205 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -54,6 +54,7 @@ class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
ParsingUtils.setPropertyValue(element, builder, "batch-time-interval");
ParsingUtils.setPropertyValue(element, builder, "disk-synchronous");
ParsingUtils.setPropertyValue(element, builder, "dispatcher-threads");
ParsingUtils.setPropertyValue(element, builder, "ignore-eviction-and-expiration");
ParsingUtils.setPropertyValue(element, builder, "maximum-queue-memory");
ParsingUtils.setPropertyValue(element, builder, "order-policy");
ParsingUtils.setPropertyValue(element, builder, "parallel");

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

@@ -39,6 +39,7 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
private Boolean batchConflationEnabled;
private Boolean diskSynchronous;
private Boolean ignoreEvictionAndExpiration;
private Boolean parallel;
private Boolean persistent;
@@ -112,6 +113,10 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
asyncEventQueueFactory.setDiskSynchronous(diskSynchronous);
}
if (ignoreEvictionAndExpiration != null) {
asyncEventQueueFactory.setIgnoreEvictionAndExpiration(ignoreEvictionAndExpiration);
}
if (maximumQueueMemory != null) {
asyncEventQueueFactory.setMaximumQueueMemory(maximumQueueMemory);
}
@@ -204,6 +209,11 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
this.dispatcherThreads = dispatcherThreads;
}
/* (non-Javadoc) */
public void setIgnoreEvictionAndExpiration(Boolean ignoreEvictionAndExpiration) {
this.ignoreEvictionAndExpiration = ignoreEvictionAndExpiration;
}
public void setMaximumQueueMemory(Integer maximumQueueMemory) {
this.maximumQueueMemory = maximumQueueMemory;
}
@@ -234,5 +244,4 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
public void setPersistent(Boolean persistent) {
this.persistent = persistent;
}
}

1
src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.8.xsd
View File

@@ -2952,6 +2952,7 @@ if an inner bean.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="ignore-eviction-and-expiration" type="xsd:string" default="true"/>
<xsd:attributeGroup ref="commonWANQueueAttributes" />
</xsd:complexType>
<!-- -->

107
src/test/java/org/springframework/data/gemfire/config/AsyncEventQueueNamespaceTest.java Normal file
View File

@@ -0,0 +1,107 @@
/*
* Copyright 2012 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.data.gemfire.config;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.junit.Assert.assertThat;
import java.util.List;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
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 AsyncEventQueueNamespaceTest class is a test suite of test cases testing the contract and functionality
* of configuring a Pivotal GemFire or Apache Geode {@link AsyncEventQueue} using the SDG XML namespace.
*
* @author John Blum
* @see org.junit.Test
* @see org.junit.runner.RunWith
* @see org.springframework.data.gemfire.config.AsyncEventQueueParser
* @see org.springframework.data.gemfire.wan.AsyncEventQueueFactoryBean
* @see org.springframework.test.context.ContextConfiguration
* @see org.springframework.test.context.junit4.SpringJUnit4ClassRunner
* @see com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueue
* @since 1.0.0
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SuppressWarnings("all")
public class AsyncEventQueueNamespaceTest {
@Autowired
private AsyncEventQueue asyncEventQueue;
@Test
public void asyncEventQueueIsConfiguredProperly() {
assertThat(asyncEventQueue, is(notNullValue(AsyncEventQueue.class)));
assertThat(asyncEventQueue.getId(), is(equalTo("TestAsyncEventQueue")));
assertThat(asyncEventQueue.isBatchConflationEnabled(), is(true));
assertThat(asyncEventQueue.getBatchSize(), is(equalTo(100)));
assertThat(asyncEventQueue.getBatchTimeInterval(), is(equalTo(30)));
assertThat(asyncEventQueue.getDiskStoreName(), is(equalTo("TestDiskStore")));
assertThat(asyncEventQueue.isDiskSynchronous(), is(true));
assertThat(asyncEventQueue.getDispatcherThreads(), is(equalTo(4)));
assertThat(asyncEventQueue.isIgnoreEvictionAndExpiration(), is(false));
assertThat(asyncEventQueue.getMaximumQueueMemory(), is(equalTo(50)));
assertThat(asyncEventQueue.getOrderPolicy(), is(equalTo(GatewaySender.OrderPolicy.KEY)));
assertThat(asyncEventQueue.isParallel(), is(false));
assertThat(asyncEventQueue.isPersistent(), is(true));
}
@Test
public void asyncEventQueueListenerEqualsExpected() {
AsyncEventListener asyncEventListener = asyncEventQueue.getAsyncEventListener();
assertThat(asyncEventListener, is(notNullValue(AsyncEventListener.class)));
assertThat(asyncEventListener.toString(), is(equalTo("TestAeqListener")));
}
public static class TestAsyncEventListener implements AsyncEventListener {
private final String name;
public TestAsyncEventListener(String name) {
this.name = name;
}
@Override
public boolean processEvents(List<AsyncEvent> events) {
return false;
}
@Override
public void close() {
}
@Override
public String toString() {
return this.name;
}
}
}

83
src/test/java/org/springframework/data/gemfire/test/StubAsyncEventQueueFactory.java
View File

@@ -38,6 +38,7 @@ public class StubAsyncEventQueueFactory implements AsyncEventQueueFactory {
private boolean batchConflationEnabled;
private boolean diskSynchronous;
private boolean ignoreEvictionAndExpiration;
private boolean parallel;
private boolean persistent;
@@ -55,63 +56,50 @@ public class StubAsyncEventQueueFactory implements AsyncEventQueueFactory {
private String diskStoreName;
@Override
public AsyncEventQueue create(final String name, final AsyncEventListener listener) {
public AsyncEventQueue create(String name, AsyncEventListener listener) {
when(asyncEventQueue.getAsyncEventListener()).thenReturn(listener);
when(asyncEventQueue.getBatchSize()).thenReturn(this.batchSize);
when(asyncEventQueue.getDiskStoreName()).thenReturn(this.diskStoreName);
when(asyncEventQueue.isPersistent()).thenReturn(this.persistent);
when(asyncEventQueue.getId()).thenReturn(name);
when(asyncEventQueue.getMaximumQueueMemory()).thenReturn(this.maxQueueMemory);
when(asyncEventQueue.isParallel()).thenReturn(this.parallel);
when(asyncEventQueue.isBatchConflationEnabled()).thenReturn(this.batchConflationEnabled);
when(asyncEventQueue.isDiskSynchronous()).thenReturn(this.diskSynchronous);
when(asyncEventQueue.getBatchSize()).thenReturn(this.batchSize);
when(asyncEventQueue.getBatchTimeInterval()).thenReturn(this.batchTimeInterval);
when(asyncEventQueue.getOrderPolicy()).thenReturn(this.orderPolicy);
when(asyncEventQueue.getDiskStoreName()).thenReturn(this.diskStoreName);
when(asyncEventQueue.isDiskSynchronous()).thenReturn(this.diskSynchronous);
when(asyncEventQueue.getDispatcherThreads()).thenReturn(this.dispatcherThreads);
when(asyncEventQueue.getGatewayEventSubstitutionFilter()).thenReturn(this.gatewayEventSubstitutionFilter);
when(asyncEventQueue.getGatewayEventFilters()).thenReturn(Collections.unmodifiableList(gatewayEventFilters));
when(asyncEventQueue.getGatewayEventSubstitutionFilter()).thenReturn(this.gatewayEventSubstitutionFilter);
when(asyncEventQueue.getId()).thenReturn(name);
when(asyncEventQueue.isIgnoreEvictionAndExpiration()).thenReturn(this.ignoreEvictionAndExpiration);
when(asyncEventQueue.getMaximumQueueMemory()).thenReturn(this.maxQueueMemory);
when(asyncEventQueue.getOrderPolicy()).thenReturn(this.orderPolicy);
when(asyncEventQueue.isParallel()).thenReturn(this.parallel);
when(asyncEventQueue.isPersistent()).thenReturn(this.persistent);
return this.asyncEventQueue;
}
public AsyncEventQueueFactory setBatchConflationEnabled(boolean batchConflationEnabled) {
this.batchConflationEnabled = batchConflationEnabled;
return this;
}
@Override
public AsyncEventQueueFactory setBatchSize(int batchSize) {
this.batchSize = batchSize;
return this;
}
public AsyncEventQueueFactory setBatchTimeInterval(int batchTimeInterval) {
this.batchTimeInterval = batchTimeInterval;
return this;
}
@Override
public AsyncEventQueueFactory setDiskStoreName(String diskStoreName) {
this.diskStoreName = diskStoreName;
return this;
}
@Override
public AsyncEventQueueFactory setMaximumQueueMemory(int maxQueueMemory) {
this.maxQueueMemory = maxQueueMemory;
return this;
}
@Override
public AsyncEventQueueFactory setPersistent(boolean persistent) {
this.persistent = persistent;
return this;
}
@Override
public AsyncEventQueueFactory setParallel(boolean parallel) {
this.parallel = parallel;
return this;
}
//The following added in 7.0.1
public AsyncEventQueueFactory setBatchConflationEnabled(boolean batchConflationEnabled) {
this.batchConflationEnabled = batchConflationEnabled;
return this;
}
public AsyncEventQueueFactory setBatchTimeInterval(int batchTimeInterval) {
this.batchTimeInterval = batchTimeInterval;
public AsyncEventQueueFactory setDispatcherThreads(int dispatchThreads) {
this.dispatcherThreads = dispatchThreads;
return this;
}
@@ -120,8 +108,13 @@ public class StubAsyncEventQueueFactory implements AsyncEventQueueFactory {
return this;
}
public AsyncEventQueueFactory setDispatcherThreads(int dispatchThreads) {
this.dispatcherThreads = dispatchThreads;
public AsyncEventQueueFactory setIgnoreEvictionAndExpiration(boolean ignoreEvictionAndExpiration) {
this.ignoreEvictionAndExpiration = ignoreEvictionAndExpiration;
return this;
}
public AsyncEventQueueFactory setMaximumQueueMemory(int maxQueueMemory) {
this.maxQueueMemory = maxQueueMemory;
return this;
}
@@ -130,22 +123,28 @@ public class StubAsyncEventQueueFactory implements AsyncEventQueueFactory {
return this;
}
@Override
public AsyncEventQueueFactory setParallel(boolean parallel) {
this.parallel = parallel;
return this;
}
public AsyncEventQueueFactory setPersistent(boolean persistent) {
this.persistent = persistent;
return this;
}
public AsyncEventQueueFactory addGatewayEventFilter(final GatewayEventFilter gatewayEventFilter) {
gatewayEventFilters.add(gatewayEventFilter);
return this;
}
@Override
public AsyncEventQueueFactory removeGatewayEventFilter(final GatewayEventFilter gatewayEventFilter) {
gatewayEventFilters.remove(gatewayEventFilter);
return this;
}
@Override
public AsyncEventQueueFactory setGatewayEventSubstitutionListener(final GatewayEventSubstitutionFilter gatewayEventSubstitutionFilter) {
this.gatewayEventSubstitutionFilter = gatewayEventSubstitutionFilter;
return this;
}
}

45
src/test/resources/org/springframework/data/gemfire/config/AsyncEventQueueNamespaceTest-context.xml Normal file
View File

@@ -0,0 +1,45 @@
<?xml version="1.0" encoding="utf-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:c="http://www.springframework.org/schema/c"
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="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
">
<bean class="org.springframework.data.gemfire.test.GemfireTestBeanPostProcessor"/>
<util:properties id="gemfireProperties">
<prop key="name">AsyncEventQueueNamespaceTest</prop>
<prop key="mcast-port">0</prop>
<prop key="log-level">warning</prop>
</util:properties>
<gfe:cache properties-ref="gemfireProperties"/>
<gfe:disk-store id="TestDiskStore">
<gfe:disk-dir location="${java.io.tmpdir}" max-size="100"/>
</gfe:disk-store>
<gfe:async-event-queue id="TestAsyncEventQueue"
batch-conflation-enabled="true"
batch-size="100"
batch-time-interval="30"
disk-store-ref="TestDiskStore"
disk-synchronous="true"
dispatcher-threads="4"
ignore-eviction-and-expiration="false"
maximum-queue-memory="50"
order-policy="KEY"
parallel="false"
persistent="true">
<gfe:async-event-listener>
<bean class="org.springframework.data.gemfire.config.AsyncEventQueueNamespaceTest.TestAsyncEventListener"
c:name="TestAeqListener"/>
</gfe:async-event-listener>
</gfe:async-event-queue>
</beans>