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

SGF-331 - Pull common WAN attributes from GatewaySenders and AsyncEventQueues in the SDG XSD into the 'commonWANQueueAttributes' attributes group.

Browse Source
This commit is contained in:
John Blum
2014-10-06 22:09:49 -07:00
parent 2896bc00f3
commit 796226d0de
8 changed files with 298 additions and 276 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -26,10 +26,15 @@ import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
/**
* Bean definition parse for the <gfe:async-event-queue> SDG XML namespace element.
*
* @author David Turanski
* @author John Blum
* @see org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser
* @see org.springframework.beans.factory.xml.ParserContext
* @see org.springframework.data.gemfire.wan.AsyncEventQueueFactoryBean
*/
public class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
@Override
protected Class<?> getBeanClass(Element element) {
@@ -40,34 +45,18 @@ public class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
builder.setLazyInit(false);
Element asyncEventListenerElement = DomUtils.getChildElementByTagName(element, "async-event-listener");
Object asyncEventListener = ParsingUtils.parseRefOrSingleNestedBeanDeclaration(parserContext,
asyncEventListenerElement, builder);
builder.addPropertyValue("asyncEventListener", asyncEventListener);
if (asyncEventListener instanceof RuntimeBeanReference) {
builder.addDependsOn(((RuntimeBeanReference) asyncEventListener).getBeanName());
}
String cacheRefAttribute = element.getAttribute("cache-ref");
String cacheName = (StringUtils.hasText(cacheRefAttribute) ? cacheRefAttribute
: GemfireConstants.DEFAULT_GEMFIRE_CACHE_NAME);
builder.addConstructorArgReference(cacheName);
parseAsyncEventListener(element, parserContext, builder);
parseCache(element, builder);
parseDiskStore(element, builder);
ParsingUtils.setPropertyValue(element, builder, "batch-size");
ParsingUtils.setPropertyValue(element, builder, "maximum-queue-memory");
ParsingUtils.setPropertyValue(element, builder, "persistent");
ParsingUtils.setPropertyValue(element, builder, "parallel");
parseDiskStore(element, builder);
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, "disk-synchronous");
ParsingUtils.setPropertyValue(element, builder, "batch-time-interval");
ParsingUtils.setPropertyValue(element, builder, "disk-synchronous");
ParsingUtils.setPropertyValue(element, builder, "dispatcher-threads");
@@ -79,15 +68,16 @@ public class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
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")) ? region.getAttribute("name")
: region.getAttribute("id");
int i = 0;
String name = regionName + ".asyncEventQueue#" + i;
String regionName = StringUtils.hasText(region.getAttribute(NAME_ATTRIBUTE))
? region.getAttribute(NAME_ATTRIBUTE) : region.getAttribute(ID_ATTRIBUTE);
int index = 0;
String name = regionName + ".asyncEventQueue#" + index;
while (parserContext.getRegistry().isBeanNameInUse(name)) {
i++;
name = regionName + ".asyncEventQueue#" + i;
name = regionName + ".asyncEventQueue#" + (++index);
}
builder.addPropertyValue("name", name);
@@ -95,6 +85,30 @@ public class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
}
}
private void parseAsyncEventListener(final Element element, final ParserContext parserContext,
final BeanDefinitionBuilder builder) {
Element asyncEventListenerElement = DomUtils.getChildElementByTagName(element, "async-event-listener");
Object asyncEventListener = ParsingUtils.parseRefOrSingleNestedBeanDeclaration(parserContext,
asyncEventListenerElement, builder);
builder.addPropertyValue("asyncEventListener", asyncEventListener);
if (asyncEventListener instanceof RuntimeBeanReference) {
builder.addDependsOn(((RuntimeBeanReference) asyncEventListener).getBeanName());
}
}
private void parseCache(final Element element, final BeanDefinitionBuilder builder) {
String cacheRefAttribute = element.getAttribute("cache-ref");
String cacheName = (StringUtils.hasText(cacheRefAttribute) ? cacheRefAttribute
: GemfireConstants.DEFAULT_GEMFIRE_CACHE_NAME);
builder.addConstructorArgReference(cacheName);
}
private void parseDiskStore(final Element element, final BeanDefinitionBuilder builder) {
ParsingUtils.setPropertyValue(element, builder, "disk-store-ref");

25
src/main/java/org/springframework/data/gemfire/config/GatewaySenderParser.java
View File

@@ -24,9 +24,13 @@ import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
/**
* Bean definition parser for the &lt;gfe:gateway-sender&gt; element in the SDG XML namespace.
*
* @author David Turanski
* @author John Blum
* @see org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser
* @see org.springframework.beans.factory.xml.ParserContext
* @see org.springframework.data.gemfire.wan.GatewaySenderFactoryBean
*/
class GatewaySenderParser extends AbstractSimpleBeanDefinitionParser {
@@ -37,19 +41,17 @@ class GatewaySenderParser extends AbstractSimpleBeanDefinitionParser {
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
String cacheRef = element.getAttribute("cache-ref");
builder.addConstructorArgReference((StringUtils.hasText(cacheRef) ? cacheRef
: GemfireConstants.DEFAULT_GEMFIRE_CACHE_NAME));
parseCache(element, builder);
ParsingUtils.setPropertyValue(element, builder, NAME_ATTRIBUTE);
ParsingUtils.setPropertyValue(element, builder, "alert-threshold");
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-store-ref");
ParsingUtils.setPropertyValue(element, builder, "disk-synchronous");
ParsingUtils.setPropertyValue(element, builder, "dispatcher-threads");
ParsingUtils.setPropertyValue(element, builder, "enable-batch-conflation");
ParsingUtils.setPropertyValue(element, builder, "manual-start");
ParsingUtils.setPropertyValue(element, builder, "maximum-queue-memory");
ParsingUtils.setPropertyValue(element, builder, "order-policy");
@@ -83,12 +85,12 @@ class GatewaySenderParser extends AbstractSimpleBeanDefinitionParser {
String regionName = (StringUtils.hasText(region.getAttribute(NAME_ATTRIBUTE))
? region.getAttribute(NAME_ATTRIBUTE) : region.getAttribute(ID_ATTRIBUTE));
int number = 0;
int index = 0;
String gatewaySenderName = (regionName + ".gatewaySender#" + number);
String gatewaySenderName = (regionName + ".gatewaySender#" + index);
while (parserContext.getRegistry().isBeanNameInUse(gatewaySenderName)) {
gatewaySenderName = (regionName + ".gatewaySender#" + (++number));
gatewaySenderName = (regionName + ".gatewaySender#" + (++index));
}
builder.addPropertyValue("name", gatewaySenderName);
@@ -96,4 +98,11 @@ class GatewaySenderParser extends AbstractSimpleBeanDefinitionParser {
}
}
private void parseCache(final Element element, final BeanDefinitionBuilder builder) {
String cacheRef = element.getAttribute("cache-ref");
builder.addConstructorArgReference((StringUtils.hasText(cacheRef) ? cacheRef
: GemfireConstants.DEFAULT_GEMFIRE_CACHE_NAME));
}
}

62
src/main/java/org/springframework/data/gemfire/config/ParsingUtils.java
View File

@@ -49,33 +49,35 @@ abstract class ParsingUtils {
private static final Log log = LogFactory.getLog(ParsingUtils.class);
static void setPropertyValue(Element element, BeanDefinitionBuilder builder, String attributeName,
String propertyName, Object defaultValue) {
String attr = element.getAttribute(attributeName);
if (StringUtils.hasText(attr)) {
builder.addPropertyValue(propertyName, attr);
} else {
if (defaultValue != null) {
builder.addPropertyValue(propertyName, defaultValue);
}
static void setPropertyValue(Element element, BeanDefinitionBuilder builder,
String attributeName, String propertyName, Object defaultValue) {
String attributeValue = element.getAttribute(attributeName);
if (StringUtils.hasText(attributeValue)) {
builder.addPropertyValue(propertyName, attributeValue);
}
else if (defaultValue != null) {
builder.addPropertyValue(propertyName, defaultValue);
}
}
static void setPropertyValue(Element element, BeanDefinitionBuilder builder, String attributeName,
String propertyName) {
setPropertyValue(element, builder, attributeName, propertyName,null);
static void setPropertyValue(Element element, BeanDefinitionBuilder builder,
String attributeName, String propertyName) {
setPropertyValue(element, builder, attributeName, propertyName, null);
}
static void setPropertyValue(Element element, BeanDefinitionBuilder builder, String attributeName) {
setPropertyValue(element, builder, attributeName, Conventions.attributeNameToPropertyName(attributeName));
}
static void setPropertyReference(Element element, BeanDefinitionBuilder builder, String attrName,
String propertyName) {
String attr = element.getAttribute(attrName);
if (StringUtils.hasText(attr)) {
builder.addPropertyReference(propertyName, attr);
static void setPropertyReference(Element element, BeanDefinitionBuilder builder,
String attributeName, String propertyName) {
String attributeValue = element.getAttribute(attributeName);
if (StringUtils.hasText(attributeValue)) {
builder.addPropertyReference(propertyName, attributeValue);
}
}
@@ -104,33 +106,25 @@ abstract class ParsingUtils {
static Object getBeanReference(ParserContext parserContext, Element element, String refAttributeName) {
String refAttributeValue = element.getAttribute(refAttributeName);
// check nested bean declarations
List<Element> childElements = DomUtils.getChildElements(element);
Object returnValue = null;
if (StringUtils.hasText(refAttributeValue)) {
if (!childElements.isEmpty()) {
if (!DomUtils.getChildElements(element).isEmpty()) {
parserContext.getReaderContext().error(String.format(
"Use either the '%1$s' attribute or a nested bean declaration for '%2$s' element, but not both.",
refAttributeName, element.getLocalName()), element);
}
return new RuntimeBeanReference(refAttributeValue);
}
else {
return null;
returnValue = new RuntimeBeanReference(refAttributeValue);
}
return returnValue;
}
static Object parseRefOrNestedCustomElement(ParserContext parserContext, Element element,
BeanDefinitionBuilder builder) {
Object beanRef = ParsingUtils.getBeanReference(parserContext, element, "bean");
if (beanRef != null) {
return beanRef;
}
else {
return parserContext.getDelegate().parseCustomElement(element, builder.getBeanDefinition());
}
return (beanRef != null ? beanRef : parserContext.getDelegate().parseCustomElement(element, builder.getBeanDefinition()));
}
static Object parseRefOrSingleNestedBeanDeclaration(ParserContext parserContext, Element element,
@@ -145,6 +139,7 @@ abstract class ParsingUtils {
static Object parseRefOrNestedBeanDeclaration(ParserContext parserContext, Element element,
BeanDefinitionBuilder builder, String refAttributeName, boolean single) {
Object beanReference = getBeanReference(parserContext, element, refAttributeName);
if (beanReference != null) {
@@ -254,6 +249,7 @@ abstract class ParsingUtils {
static void parseTransportFilters(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
Element transportFilterElement = DomUtils.getChildElementByTagName(element, "transport-filter");
if (transportFilterElement != null) {
builder.addPropertyValue("transportFilters", parseRefOrNestedBeanDeclaration(parserContext,
transportFilterElement, builder));

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

@@ -30,6 +30,7 @@ import com.gemstone.gemfire.cache.util.Gateway;
* @author David Turanski
* @author John Blum
*/
@SuppressWarnings("unused")
public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<AsyncEventQueue> {
private AsyncEventListener asyncEventListener;
@@ -150,20 +151,72 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
this.asyncEventListener = listener;
}
public void setDiskStoreRef(String diskStoreRef) {
this.diskStoreReference = diskStoreRef;
/**
* @param asyncEventQueue overrides Async Event Queue returned by this FactoryBean.
*/
public void setAsyncEventQueue(AsyncEventQueue asyncEventQueue) {
this.asyncEventQueue = asyncEventQueue;
}
/**
* Enable or disable the Async Event Queue's (AEQ) should conflate messages.
*
* @param batchConflationEnabled a boolean value indicating whether to conflate queued events.
*/
public void setBatchConflationEnabled(Boolean batchConflationEnabled) {
this.batchConflationEnabled = batchConflationEnabled;
}
public void setBatchSize(Integer batchSize) {
this.batchSize = batchSize;
}
/**
* Set the Aysync Event Queue's (AEQ) interval between sending batches.
*
* @param batchTimeInterval an integer value indicating the maximum number of milliseconds that can elapse
* between sending batches.
*/
public void setBatchTimeInterval(Integer batchTimeInterval) {
this.batchTimeInterval = batchTimeInterval;
}
public void setDiskStoreRef(String diskStoreRef) {
this.diskStoreReference = diskStoreRef;
}
/**
* Set the Async Event Queue (AEQ) disk write synchronization policy.
*
* @param diskSynchronous a boolean value indicating whether disk writes are synchronous.
*/
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).
*
* @param dispatcherThreads an Integer indicating the number of dispatcher threads used to process Region Events
* from the associated Queue.
*/
public void setDispatcherThreads(Integer dispatcherThreads) {
this.dispatcherThreads = dispatcherThreads;
}
public void setMaximumQueueMemory(Integer maximumQueueMemory) {
this.maximumQueueMemory = maximumQueueMemory;
}
public void setPersistent(Boolean persistent) {
this.persistent = persistent;
/**
* 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.
*
* @param orderPolicy a String to indicate the AEQ order policy.
*/
public void setOrderPolicy(String orderPolicy) {
this.orderPolicy = orderPolicy;
}
public void setParallel(Boolean parallel) {
@@ -178,60 +231,8 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
return Boolean.TRUE.equals(parallel);
}
/**
* @param asyncEventQueue overrides Async Event Queue returned by this FactoryBean.
*/
public void setAsyncEventQueue(AsyncEventQueue asyncEventQueue) {
this.asyncEventQueue = asyncEventQueue;
}
/**
* Set the number of dispatcher threads used to process Region events from the associated Aysnc Event Queue (AEQ).
*
* @param dispatcherThreads an integer indicating the number of dispatcher threads used to process Region events
* from the associated queue.
*/
public void setDispatcherThreads(Integer dispatcherThreads) {
this.dispatcherThreads = dispatcherThreads;
}
/**
* Set the Aysync Event Queue's (AEQ) interval between sending batches.
*
* @param batchTimeInterval an integer value indicating the maximum number of milliseconds that can elapse
* between sending batches.
*/
public void setBatchTimeInterval(Integer batchTimeInterval) {
this.batchTimeInterval = batchTimeInterval;
}
/**
* Enable or disable the Async Event Queue's (AEQ) should conflate messages.
*
* @param batchConflationEnabled a boolean value indicating whether to conflate queued events.
*/
public void setBatchConflationEnabled(Boolean batchConflationEnabled) {
this.batchConflationEnabled = batchConflationEnabled;
}
/**
* Set the Async Event Queue (AEQ) disk write synchronization policy.
*
* @param diskSynchronous a boolean value indicating whether disk writes are synchronous.
*/
public void setDiskSynchronous(Boolean diskSynchronous) {
this.diskSynchronous = diskSynchronous;
}
/**
* 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.
*
* @param orderPolicy a String to indicate the AEQ order policy.
*/
public void setOrderPolicy(String orderPolicy) {
this.orderPolicy = orderPolicy;
public void setPersistent(Boolean persistent) {
this.persistent = persistent;
}
}

48
src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java
View File

@@ -57,7 +57,7 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
private List<GatewayTransportFilter> transportFilters;
private Boolean diskSynchronous;
private Boolean enableBatchConflation;
private Boolean batchConflationEnabled;
private Boolean parallel;
private Boolean persistent;
@@ -96,8 +96,8 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
@Override
protected void doInit() {
GatewaySenderFactory gatewaySenderFactory = (this.factory != null ? (GatewaySenderFactory) factory :
cache.createGatewaySenderFactory());
GatewaySenderFactory gatewaySenderFactory = (this.factory != null ? (GatewaySenderFactory) factory
: cache.createGatewaySenderFactory());
if (alertThreshold != null) {
gatewaySenderFactory.setAlertThreshold(alertThreshold);
@@ -123,8 +123,8 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
gatewaySenderFactory.setDispatcherThreads(dispatcherThreads);
}
if (enableBatchConflation != null) {
gatewaySenderFactory.setBatchConflationEnabled(enableBatchConflation);
if (batchConflationEnabled != null) {
gatewaySenderFactory.setBatchConflationEnabled(batchConflationEnabled);
}
if (!CollectionUtils.isEmpty(eventFilters)) {
@@ -176,24 +176,24 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
gatewaySender = wrapper;
}
public void setRemoteDistributedSystemId(int remoteDistributedSystemId) {
this.remoteDistributedSystemId = remoteDistributedSystemId;
}
public void setEventFilters(List<GatewayEventFilter> gatewayEventFilters) {
this.eventFilters = gatewayEventFilters;
}
public void setTransportFilters(List<GatewayTransportFilter> gatewayTransportFilters) {
this.transportFilters = gatewayTransportFilters;
}
public void setAlertThreshold(Integer alertThreshold) {
this.alertThreshold = alertThreshold;
}
public void setBatchConflationEnabled(Boolean batchConflationEnabled) {
this.batchConflationEnabled = batchConflationEnabled;
}
/**
* Boolean value that determines whether GemFire should conflate messages.
*
* @param enableBatchConflation a boolean value indicating whether GemFire should conflate messages in the Queue.
* @see #setBatchConflationEnabled(Boolean)
* @deprecated use setBatchConflationEnabled(Boolean)
*/
@Deprecated
public void setEnableBatchConflation(Boolean enableBatchConflation) {
this.enableBatchConflation = enableBatchConflation;
this.batchConflationEnabled = enableBatchConflation;
}
public void setBatchSize(Integer batchSize) {
@@ -216,6 +216,10 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
this.dispatcherThreads = dispatcherThreads;
}
public void setEventFilters(List<GatewayEventFilter> gatewayEventFilters) {
this.eventFilters = gatewayEventFilters;
}
public void setEventSubstitutionFilter(final GatewayEventSubstitutionFilter eventSubstitutionFilter) {
this.eventSubstitutionFilter = eventSubstitutionFilter;
}
@@ -256,6 +260,10 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
return Boolean.TRUE.equals(this.persistent);
}
public void setRemoteDistributedSystemId(int remoteDistributedSystemId) {
this.remoteDistributedSystemId = remoteDistributedSystemId;
}
public void setSocketBufferSize(Integer socketBufferSize) {
this.socketBufferSize = socketBufferSize;
}
@@ -264,6 +272,10 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
this.socketReadTimeout = socketReadTimeout;
}
public void setTransportFilters(List<GatewayTransportFilter> gatewayTransportFilters) {
this.transportFilters = gatewayTransportFilters;
}
/* (non-Javadoc)
* @see org.springframework.context.SmartLifecycle#isAutoStartup()
*/

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

@@ -2576,36 +2576,99 @@ A gateway sender gateway definition (requires Gemfire 7.0 or later)
<xsd:element name="event-substitution-filter" type="gatewayEventSubstitutionFilterType" minOccurs="0" maxOccurs="1"/>
<xsd:element name="transport-filter" type="gatewayTransportFilterType" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attributeGroup ref="commonWANQueueAttributes" />
<xsd:attribute name="name" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Optionally specifies the GemFire gateway sender id. By default this value is the bean id or a generated value if an inner bean.
Optionally specifies the GemFire GatewaySender id. By default, this value is the bean id or a generated value
if an inner bean.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="remote-distributed-system-id" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the remote distributed system id, an integer value representing the remote distributed system
Integer that uniquely identifies the remote GemFire cluster to which this GatewaySender will send Region Events.
This value corresponds to the distributed-system-id property specified in Locators for the remote cluster.
This attribute is required.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="alert-threshold" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the alert threshold in miliseconds, indicating the maximum time elapsed from when the gateway sent the message
to when the acknowldgement was received from the gateway receiver.
Maximum number of milliseconds that a Region Event can remain in the GatewaySender Queue before GemFire logs an alert.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="enable-batch-conflation" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Boolean value that determines whether GemFire should conflate messages. GemFire default is false.
NOTE, this attribute is deprecated in favor of the common WAN Queue attribute, 'batch-conflation-enabled'. If both
attributes are specified, then 'batch-conflation-enabled' takes precedence.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="manual-start" type="xsd:string" default="false">
<xsd:annotation>
<xsd:documentation><![CDATA[
Boolean value that specifies whether you need to manually start the GatewaySender. If you supply a null value,
the default is "false" and the GatewaySender attempts to start automatically.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="socket-buffer-size" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Size of the socket buffer that sends messages to remote sites. This size should match the size of the
socket-buffer-size attribute of remote GatewayReceivers that process Region Events.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="socket-read-timeout" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Amount of time in milliseconds that the GatewaySender will wait to receive an acknowledgment from a remote site.
By default this is set to 0, which means there is no timeout. If you do set this timeout, you must set it to
a minimum of 30000 (milliseconds). Setting it to a lower number will generate an error message and reset the value
to the default of 0.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="commonWANQueueAttributes"/>
</xsd:complexType>
<!-- -->
<xsd:attributeGroup name="commonWANQueueAttributes">
<xsd:attribute name="batch-conflation-enabled" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Boolean value that determines whether GemFire should conflate messages. GemFire default is false.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="batch-size" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Maximum number of messages that a batch can contain.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="batch-time-interval" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The maximum time interval that can elapse before a partial batch is sent from a GatewaySender to its corresponding GatewayReceiver.
Maximum number of milliseconds that can elapse between sending batches.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="disk-store-ref" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Named DiskStore to use for storing the Queue overflow, or for persisting the Queue. If you specify a value,
the named DiskStore must exist. If you specify a null value, GemFire uses the Default DiskStore for overflow
and Queue persistence.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="disk-synchronous" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -2617,84 +2680,51 @@ for the Region or asynchronously (false).
<xsd:attribute name="dispatcher-threads" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the number of dispatcher threads to allocate to the gateway sender
Number of dispatcher threads that are used to process Region Events from a GatewaySender Queue
or Asynchronous Event Queue.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="enable-batch-conflation" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies whether batch conflation is enabled (true or false)
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="manual-start" type="xsd:string" default="false">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies if the gateway sender is manually (true) or automatically (false) started. Default is an automatic start (false).
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order-policy" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the order policy. Order policy only applies if the gateway sender queue is serial:
KEY: Indicates that events will be concurrently processed based on the event's key,
PARTITION: Indicates that events will be concurrently processed based on the event's partition (using the PartitionResolver)
THREAD: Indicates that events will be concurrently processed based on the event's originating member and thread
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="socket-buffer-size" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the socket buffer size in bytes
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="socket-read-timeout" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the socket read timeout in milliseconds
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<!-- -->
<xsd:attributeGroup name="commonWANQueueAttributes">
<xsd:attribute name="batch-size" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the batch size
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="disk-store-ref" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Indicates the id of disk store to use for persistence
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="maximum-queue-memory" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the maximum memory in MB to allocate for the queue
Maximum amount of memory in megabytes that the Queue can consume before overflowing to disk.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order-policy" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
When the dispatcher-threads attribute is greater than 1, order-policy configures the way in which multiple
dispatcher threads process Region Events from a serial Gateway Queue or serial Asynchronous Event Queue.
This attribute can have one of the following values:
KEY - When distributing Region Events from the local Queue, multiple dispatcher threads preserve the order of key updates.
THREAD - When distributing Region Events from the local Queue, multiple dispatcher threads preserve the order
in which a given thread added Region Events to the Queue.
PARTITION - When distributing Region Events from the local Queue, multiple dispatcher threads preserve the order
in which Region Events were added to the local Queue. For a Partitioned Region, this means that all Region Events
delivered to a specific partition are delivered in the same order to the remote GemFire site. For a Distributed Region,
this means that all key updates delivered to the local GatewaySender Queue are distributed to the remote site
in the same order.
You cannot configure the order-policy for a parallel Event Queue, because parallel Queues cannot preserve event
ordering for Regions. Only the ordering of events for a given partition (or in a given Queue of a Distributed Region)
can be preserved.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="parallel" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
A value of "false" or "true" that specifies the type of queue GemFire creates, serial or parallel.
]]></xsd:documentation>
Value of "true" or "false" specifying the type of GatewaySender or AsyncEventQueue that GemFire creates.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="persistent" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies whether persistence is enabled: true or false(default)
Boolean value that determines whether GemFire persists the Gateway Queue or AsyncEventQueue.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -2795,16 +2825,15 @@ An async event queue definition (requires Gemfire 7.0 or later)
</xsd:appinfo>
</xsd:annotation>
<xsd:sequence>
<xsd:element name="async-event-listener" minOccurs="1"
maxOccurs="1">
<xsd:element name="async-event-listener" minOccurs="1" maxOccurs="1">
<xsd:annotation>
<xsd:documentation
source="com.gemstone.gemfire.cache.wan.AsyncEventListener"><![CDATA[
An async event listener definition for this distributed system. (requires Gemfire 7.0)
An AsyncEventListener bean definition for this AsyncEventQueue. (requires Gemfire 7.0)
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:exports type="com.gemstone.gemfire.cache.wan.AsyncEventListener" />
<tool:exports type="com.gemstone.gemfire.cache.wan.AsyncEventListener"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
@@ -2833,56 +2862,11 @@ use inner bean declarations.
<xsd:attribute name="name" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Optionally specifies the GemFire async event queue id. By default this value is the bean id or a generated value if an inner bean.
Optionally specifies the GemFire AsyncEventQueue id. By default this value is the bean id or a generated value
if an inner bean.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="batch-conflation-enabled" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Optionally specifies whether to conflate queued events (true or false)
(only available in Gemfire 7.0.1 + )
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="batch-time-interval" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the maximum time interval that can elapse before a partial batch is sent from a the AsyncEventQueue
(only available in Gemfire 7.0.1 + )
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="disk-synchronous" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies if disk writes should be synchronous (true or false) (only available in Gemfire 7.0.1 + )
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="dispatcher-threads" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the number of dispatcher threads to use (only available in Gemfire 7.0.1 + )
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order-policy" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
(Only available in GemFire 7.0.1 + )
Specifies the order policy - This only applies if parallel is enabled:
KEY: Indicates that events will be parallelized based on the event's key,
PARTITION:Indicates that events will be parallelized based on the event's: partition (using the PartitionResolver)
THREAD:Indicates that events will be parallelized based on the event's originating member and thread
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="commonWANQueueAttributes" />
</xsd:complexType>
<!-- -->

24
src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java
View File

@@ -93,16 +93,18 @@ public class GemfireV7GatewayNamespaceTest extends RecreatingContextTest {
@Test
public void testAsyncEventQueue() {
AsyncEventQueue aseq = ctx.getBean("async-event-queue", AsyncEventQueue.class);
assertEquals(10, aseq.getBatchSize());
assertTrue(aseq.isPersistent());
assertFalse(aseq.isParallel());
assertEquals("diskstore", aseq.getDiskStoreName());
assertEquals(50, aseq.getMaximumQueueMemory());
assertEquals(3, aseq.getBatchTimeInterval());
assertEquals(OrderPolicy.KEY, aseq.getOrderPolicy());
assertEquals(true, aseq.isDiskSynchronous());
assertEquals(true, aseq.isBatchConflationEnabled());
AsyncEventQueue asyncEventQueue = ctx.getBean("async-event-queue", AsyncEventQueue.class);
assertNotNull(asyncEventQueue);
assertTrue(asyncEventQueue.isBatchConflationEnabled());
assertEquals(10, asyncEventQueue.getBatchSize());
assertEquals(3, asyncEventQueue.getBatchTimeInterval());
assertEquals("diskstore", asyncEventQueue.getDiskStoreName());
assertTrue(asyncEventQueue.isDiskSynchronous());
assertEquals(50, asyncEventQueue.getMaximumQueueMemory());
assertEquals(OrderPolicy.KEY, asyncEventQueue.getOrderPolicy());
assertFalse(asyncEventQueue.isParallel());
assertTrue(asyncEventQueue.isPersistent());
}
@Test
@@ -113,6 +115,7 @@ public class GemfireV7GatewayNamespaceTest extends RecreatingContextTest {
assertNotNull(TestUtils.readField("cache", gatewaySenderFactoryBean));
assertEquals(2, TestUtils.readField("remoteDistributedSystemId", gatewaySenderFactoryBean));
assertEquals(10, TestUtils.readField("alertThreshold", gatewaySenderFactoryBean));
assertTrue(Boolean.TRUE.equals(TestUtils.readField("batchConflationEnabled", gatewaySenderFactoryBean)));
assertEquals(11, TestUtils.readField("batchSize", gatewaySenderFactoryBean));
assertEquals(12, TestUtils.readField("dispatcherThreads", gatewaySenderFactoryBean));
assertEquals(false, TestUtils.readField("diskSynchronous", gatewaySenderFactoryBean));
@@ -187,6 +190,7 @@ public class GemfireV7GatewayNamespaceTest extends RecreatingContextTest {
assertNotNull(gatewaySenderFactoryBean);
assertNotNull(TestUtils.readField("cache", gatewaySenderFactoryBean));
assertEquals(3, TestUtils.readField("remoteDistributedSystemId", gatewaySenderFactoryBean));
assertTrue(Boolean.TRUE.equals(TestUtils.readField("batchConflationEnabled", gatewaySenderFactoryBean)));
assertEquals(50, TestUtils.readField("batchSize", gatewaySenderFactoryBean));
assertEquals(10, TestUtils.readField("dispatcherThreads", gatewaySenderFactoryBean));
assertEquals(true, TestUtils.readField("manualStart", gatewaySenderFactoryBean));

8
src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml
View File

@@ -69,8 +69,9 @@
</gfe:gateway-receiver>
<!-- need manual-start=true for the unit test because GF will throw an exception if no locators are configured -->
<gfe:gateway-sender id="gateway-sender" remote-distributed-system-id="2" alert-threshold="10" batch-size="11"
dispatcher-threads="12" disk-synchronous="false" manual-start="true">
<gfe:gateway-sender id="gateway-sender" remote-distributed-system-id="2" alert-threshold="10"
batch-conflation-enabled="true" batch-size="11" dispatcher-threads="12" disk-synchronous="false"
manual-start="true">
<gfe:event-filter>
<ref bean="event-filter"/>
<bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestEventFilter"/>
@@ -82,7 +83,8 @@
</gfe:gateway-sender>
<gfe:gateway-sender id="gateway-sender-with-event-transport-filter-refs" remote-distributed-system-id="3"
batch-size="50" dispatcher-threads="10" manual-start="true">
batch-conflation-enabled="true" batch-size="50" dispatcher-threads="10" enable-batch-conflation="false"
manual-start="true">
<gfe:event-filter ref="event-filter"/>
<gfe:transport-filter ref="transport-filter"/>
</gfe:gateway-sender>