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

DATAGEODE-91 - Impossible to define event filter for AsyncEventQueue.

Browse Source
This commit is contained in:
John Blum
2018-03-22 17:58:19 -07:00
parent 781813448f
commit 867ad2062f
9 changed files with 620 additions and 341 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -62,19 +62,37 @@ 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, "foward-expiration-destroy");
ParsingUtils.setPropertyValue(element, builder, "forward-expiration-destroy");
ParsingUtils.setPropertyValue(element, builder, "maximum-queue-memory");
ParsingUtils.setPropertyValue(element, builder, "order-policy");
ParsingUtils.setPropertyValue(element, builder, "parallel");
ParsingUtils.setPropertyValue(element, builder, "persistent");
Element eventFilterElement = DomUtils.getChildElementByTagName(element, "event-filter");
if (eventFilterElement != null) {
builder.addPropertyValue("gatewayEventFilters",
ParsingUtils.parseRefOrNestedBeanDeclaration(eventFilterElement, parserContext, builder));
}
Element eventSubstitutionFilterElement =
DomUtils.getChildElementByTagName(element, "event-substitution-filter");
if (eventSubstitutionFilterElement != null) {
builder.addPropertyValue("gatewayEventSubstitutionFilter",
ParsingUtils.parseRefOrSingleNestedBeanDeclaration(eventSubstitutionFilterElement, parserContext, 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))
? region.getAttribute(NAME_ATTRIBUTE) : region.getAttribute(ID_ATTRIBUTE);
? region.getAttribute(NAME_ATTRIBUTE)
: region.getAttribute(ID_ATTRIBUTE);
int index = 0;
@@ -91,11 +109,11 @@ class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
/* (non-Javadoc) */
private void parseAsyncEventListener(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
Element asyncEventListenerElement = DomUtils.getChildElementByTagName(element, "async-event-listener");
Object asyncEventListener = ParsingUtils.parseRefOrSingleNestedBeanDeclaration(asyncEventListenerElement,
parserContext,
builder);
Object asyncEventListener =
ParsingUtils.parseRefOrSingleNestedBeanDeclaration(asyncEventListenerElement, parserContext, builder);
builder.addPropertyValue("asyncEventListener", asyncEventListener);
@@ -106,15 +124,16 @@ class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
/* (non-Javadoc) */
private void parseCache(Element element, BeanDefinitionBuilder builder) {
String cacheRefAttribute = element.getAttribute("cache-ref");
String cacheName = SpringUtils.defaultIfEmpty(cacheRefAttribute,
GemfireConstants.DEFAULT_GEMFIRE_CACHE_NAME);
String cacheName = SpringUtils.defaultIfEmpty(cacheRefAttribute, GemfireConstants.DEFAULT_GEMFIRE_CACHE_NAME);
builder.addConstructorArgReference(cacheName);
}
/* (non-Javadoc) */
private void parseDiskStore(Element element, BeanDefinitionBuilder builder) {
ParsingUtils.setPropertyValue(element, builder, "disk-store-ref");
String diskStoreRef = element.getAttribute("disk-store-ref");

40
src/main/java/org/springframework/data/gemfire/wan/AbstractWANComponentFactoryBean.java
View File

@@ -13,10 +13,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.gemfire.wan;
import java.util.Arrays;
import java.util.List;
package org.springframework.data.gemfire.wan;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -29,35 +27,34 @@ import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Base class for GemFire WAN Gateway component factory beans.
* Abstract base class for WAN Gateway components.
*
* @author David Turanski
* @author John Blum
* @see org.apache.geode.cache.Cache
* @see org.springframework.beans.factory.BeanNameAware
* @see org.springframework.beans.factory.DisposableBean
* @see org.springframework.beans.factory.FactoryBean
* @see org.springframework.beans.factory.InitializingBean
*/
public abstract class AbstractWANComponentFactoryBean<T> implements BeanNameAware, FactoryBean<T>,
InitializingBean, DisposableBean {
protected static final List<String> VALID_ORDER_POLICIES = Arrays.asList("KEY", "PARTITION", "THREAD");
protected Log log = LogFactory.getLog(getClass());
public abstract class AbstractWANComponentFactoryBean<T>
implements BeanNameAware, DisposableBean, FactoryBean<T>, InitializingBean {
protected final Cache cache;
protected final Log log = LogFactory.getLog(getClass());
protected Object factory;
private String beanName;
private String name;
protected AbstractWANComponentFactoryBean(final Cache cache) {
protected AbstractWANComponentFactoryBean(Cache cache) {
this.cache = cache;
}
@Override
public final void setBeanName(final String beanName) {
public void setBeanName(String beanName) {
this.beanName = beanName;
}
@@ -65,20 +62,14 @@ public abstract class AbstractWANComponentFactoryBean<T> implements BeanNameAwar
this.factory = factory;
}
public void setName(final String name) {
public void setName(String name) {
this.name = name;
}
public String getName() {
return (StringUtils.hasText(name) ? name : beanName);
return StringUtils.hasText(this.name) ? this.name : this.beanName;
}
@Override
public abstract T getObject() throws Exception;
@Override
public abstract Class<?> getObjectType();
@Override
public final boolean isSingleton() {
return true;
@@ -86,15 +77,16 @@ public abstract class AbstractWANComponentFactoryBean<T> implements BeanNameAwar
@Override
public final void afterPropertiesSet() throws Exception {
Assert.notNull(cache, "Cache must not be null.");
Assert.notNull(getName(), "Name must not be null.");
Assert.notNull(this.cache, "Cache must not be null");
Assert.notNull(getName(), "Name must not be null");
doInit();
}
protected abstract void doInit() throws Exception;
@Override
public void destroy() throws Exception {
}
public void destroy() throws Exception { }
}

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

@@ -13,21 +13,36 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.gemfire.wan;
import static org.springframework.data.gemfire.util.CollectionUtils.nullSafeList;
import java.util.List;
import java.util.Optional;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.asyncqueue.AsyncEventListener;
import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory;
import org.apache.geode.cache.wan.GatewayEventFilter;
import org.apache.geode.cache.wan.GatewayEventSubstitutionFilter;
import org.apache.geode.cache.wan.GatewaySender;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.util.Assert;
/**
* FactoryBean for creating GemFire {@link AsyncEventQueue}s.
* Spring {@link FactoryBean} for creating Apache Geode/Pivotal GemFire {@link AsyncEventQueue AsyncEventQueues}.
*
* @author David Turanski
* @author John Blum
* @see org.apache.geode.cache.Cache
* @see org.apache.geode.cache.Region
* @see org.apache.geode.cache.asyncqueue.AsyncEventListener
* @see org.apache.geode.cache.asyncqueue.AsyncEventQueue
* @see org.springframework.data.gemfire.wan.AbstractWANComponentFactoryBean
*/
@SuppressWarnings("unused")
public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<AsyncEventQueue> {
@@ -47,8 +62,13 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
private Integer dispatcherThreads;
private Integer maximumQueueMemory;
private GatewayEventSubstitutionFilter gatewayEventSubstitutionFilter;
private GatewaySender.OrderPolicy orderPolicy;
private List<GatewayEventFilter> gatewayEventFilters;
private String diskStoreReference;
private String orderPolicy;
/**
* Constructs an instance of the AsyncEventQueueFactoryBean for creating an GemFire AsyncEventQueue.
@@ -67,81 +87,59 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
* @param asyncEventListener required {@link 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 != null ? asyncEventQueue.getClass() : AsyncEventQueue.class);
return this.asyncEventQueue != null ? this.asyncEventQueue.getClass() : AsyncEventQueue.class;
}
@Override
protected void doInit() {
Assert.notNull(this.asyncEventListener, "AsyncEventListener must not be null");
Assert.state(this.asyncEventListener != null, "AsyncEventListener must not 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);
}
if (batchTimeInterval != null) {
asyncEventQueueFactory.setBatchTimeInterval(batchTimeInterval);
}
if (batchConflationEnabled != null) {
asyncEventQueueFactory.setBatchConflationEnabled(batchConflationEnabled);
}
if (dispatcherThreads != null) {
asyncEventQueueFactory.setDispatcherThreads(dispatcherThreads);
}
if (diskStoreReference != null) {
asyncEventQueueFactory.setDiskStoreName(diskStoreReference);
}
if (diskSynchronous != null) {
asyncEventQueueFactory.setDiskSynchronous(diskSynchronous);
}
if (forwardExpirationDestroy != null) {
asyncEventQueueFactory.setForwardExpirationDestroy(forwardExpirationDestroy);
}
if (maximumQueueMemory != null) {
asyncEventQueueFactory.setMaximumQueueMemory(maximumQueueMemory);
}
Optional.ofNullable(this.batchConflationEnabled).ifPresent(asyncEventQueueFactory::setBatchConflationEnabled);
Optional.ofNullable(this.batchSize).ifPresent(asyncEventQueueFactory::setBatchSize);
Optional.ofNullable(this.batchTimeInterval).ifPresent(asyncEventQueueFactory::setBatchTimeInterval);
Optional.ofNullable(this.diskStoreReference).ifPresent(asyncEventQueueFactory::setDiskStoreName);
Optional.ofNullable(this.diskSynchronous).ifPresent(asyncEventQueueFactory::setDiskSynchronous);
Optional.ofNullable(this.dispatcherThreads).ifPresent(asyncEventQueueFactory::setDispatcherThreads);
Optional.ofNullable(this.forwardExpirationDestroy).ifPresent(asyncEventQueueFactory::setForwardExpirationDestroy);
Optional.ofNullable(this.gatewayEventSubstitutionFilter).ifPresent(asyncEventQueueFactory::setGatewayEventSubstitutionListener);
Optional.ofNullable(this.maximumQueueMemory).ifPresent(asyncEventQueueFactory::setMaximumQueueMemory);
Optional.ofNullable(this.persistent).ifPresent(asyncEventQueueFactory::setPersistent);
asyncEventQueueFactory.setParallel(isParallelEventQueue());
if (orderPolicy != null) {
Assert.isTrue(isSerialEventQueue(), "Order Policy cannot be used with a Parallel Event Queue");
nullSafeList(this.gatewayEventFilters).forEach(asyncEventQueueFactory::addGatewayEventFilter);
Assert.isTrue(VALID_ORDER_POLICIES.contains(orderPolicy.toUpperCase()), String.format(
"The value of Order Policy '$1%s' is invalid", orderPolicy));
if (this.orderPolicy != null) {
asyncEventQueueFactory.setOrderPolicy(GatewaySender.OrderPolicy.valueOf(orderPolicy.toUpperCase()));
Assert.state(isSerialEventQueue(), "OrderPolicy cannot be used with a Parallel AsyncEventQueue");
asyncEventQueueFactory.setOrderPolicy(this.orderPolicy);
}
if (persistent != null) {
asyncEventQueueFactory.setPersistent(persistent);
}
asyncEventQueue = asyncEventQueueFactory.create(getName(), this.asyncEventListener);
setAsyncEventQueue(asyncEventQueueFactory.create(getName(), this.asyncEventListener));
}
@Override
public void destroy() throws Exception {
if (!cache.isClosed()) {
if (!this.cache.isClosed()) {
try {
this.asyncEventListener.close();
}
@@ -151,6 +149,7 @@ 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");
@@ -158,16 +157,20 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
}
/**
* @param asyncEventQueue overrides Async Event Queue returned by this FactoryBean.
* Configures the {@link AsyncEventQueue} returned by this {@link FactoryBean}.
*
* @param asyncEventQueue overrides {@link AsyncEventQueue} returned by this {@link FactoryBean}.
* @see org.apache.geode.cache.asyncqueue.AsyncEventQueue
*/
public void setAsyncEventQueue(AsyncEventQueue asyncEventQueue) {
this.asyncEventQueue = asyncEventQueue;
}
/**
* Enable or disable the Async Event Queue's (AEQ) should conflate messages.
* Enable or disable {@link AsyncEventQueue} (AEQ) message conflation.
*
* @param batchConflationEnabled a boolean value indicating whether to conflate queued events.
* @param batchConflationEnabled {@link Boolean} indicating whether to conflate queued events.
* @see org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory#setBatchConflationEnabled(boolean)
*/
public void setBatchConflationEnabled(Boolean batchConflationEnabled) {
this.batchConflationEnabled = batchConflationEnabled;
@@ -178,10 +181,11 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
}
/**
* Set the Aysync Event Queue's (AEQ) interval between sending batches.
* Configures the {@link AsyncEventQueue} (AEQ) interval between sending batches.
*
* @param batchTimeInterval an integer value indicating the maximum number of milliseconds that can elapse
* between sending batches.
* @param batchTimeInterval {@link Integer} specifying the maximum number of milliseconds
* that can elapse between sending batches.
* @see org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory#setBatchTimeInterval(int)
*/
public void setBatchTimeInterval(Integer batchTimeInterval) {
this.batchTimeInterval = batchTimeInterval;
@@ -192,19 +196,22 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
}
/**
* Set the Async Event Queue (AEQ) disk write synchronization policy.
* Configures the {@link AsyncEventQueue} (AEQ) disk write synchronization policy.
*
* @param diskSynchronous a boolean value indicating whether disk writes are synchronous.
* @param diskSynchronous boolean value indicating whether disk writes are synchronous.
* @see org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory#setDiskSynchronous(boolean)
*/
public void setDiskSynchronous(Boolean diskSynchronous) {
this.diskSynchronous = diskSynchronous;
}
/**
* Set the number of dispatcher threads used to process Region Events from the associated Async Event Queue (AEQ).
* Configures the number of dispatcher threads used to process Region Events
* from the associated {@link AsyncEventQueue} (AEQ).
*
* @param dispatcherThreads an Integer indicating the number of dispatcher threads used to process Region Events
* from the associated Queue.
* @param dispatcherThreads {@link Integer} specifying the number of dispatcher threads used
* to process {@link Region} events from the associated queue.
* @see org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory#setDispatcherThreads(int)
*/
public void setDispatcherThreads(Integer dispatcherThreads) {
this.dispatcherThreads = dispatcherThreads;
@@ -213,8 +220,8 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
/**
* Forwards expiration (action-based) destroy events to the {@link AsyncEventQueue} (AEQ).
*
* By default, destroy events are not added to the AEQ. Setting this attribute to
* {@literal true} will add all expiration destroy events to the AEQ.
* By default, destroy events are not added to the AEQ. Setting this attribute to {@literal true}
* will add all expiration destroy events to the AEQ.
*
* @param forwardExpirationDestroy boolean value indicating whether to forward expiration destroy events.
* @see org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory#setForwardExpirationDestroy(boolean)
@@ -225,18 +232,33 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
this.forwardExpirationDestroy = forwardExpirationDestroy;
}
public void setGatewayEventFilters(List<GatewayEventFilter> eventFilters) {
this.gatewayEventFilters = eventFilters;
}
public void setGatewayEventSubstitutionFilter(GatewayEventSubstitutionFilter eventSubstitutionFilter) {
this.gatewayEventSubstitutionFilter = eventSubstitutionFilter;
}
public void setMaximumQueueMemory(Integer maximumQueueMemory) {
this.maximumQueueMemory = maximumQueueMemory;
}
/**
* Set the Async Event Queue (AEQ) ordering policy (e.g. KEY, PARTITION, THREAD). When dispatcher threads
* are greater than 1, the ordering policy configures the way in which multiple dispatcher threads
* process Region events from the queue.
* Configures the {@link AsyncEventQueue} (AEQ) ordering policy (e.g. {@literal KEY}, {@literal PARTITION},
* {@literal THREAD}).
*
* @param orderPolicy a String to indicate the AEQ order policy.
* When dispatcher threads are greater than one, the ordering policy configures the way in which
* multiple dispatcher threads process Region events from the queue.
*
* @param orderPolicy {@link String} specifying the name of the AEQ order policy.
* @see org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory#setOrderPolicy(GatewaySender.OrderPolicy)
*/
public void setOrderPolicy(String orderPolicy) {
setOrderPolicy(GatewaySender.OrderPolicy.valueOf(String.valueOf(orderPolicy).toUpperCase()));
}
public void setOrderPolicy(GatewaySender.OrderPolicy orderPolicy) {
this.orderPolicy = orderPolicy;
}

5
src/main/resources/org/springframework/data/gemfire/config/spring-geode-1.1.xsd
View File

@@ -2931,8 +2931,7 @@ An AsyncEventListener bean definition for this AsyncEventQueue. (requires Gemfir
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:any namespace="##other" processContents="skip"
minOccurs="0" maxOccurs="unbounded">
<xsd:any namespace="##other" processContents="skip" minOccurs="0" maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation><![CDATA[
Inner bean definition of the async event listener
@@ -2950,6 +2949,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>

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

@@ -2971,6 +2971,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>