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 17:58:19 -07:00
parent 435130b67b
commit cceff39f72
11 changed files with 842 additions and 393 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
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,34 +49,50 @@ 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, "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");
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(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;
@@ -92,13 +107,12 @@ 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);
@@ -109,15 +123,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 { }
}

167
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> {
@@ -38,6 +53,7 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
private Boolean batchConflationEnabled;
private Boolean diskSynchronous;
private Boolean forwardExpirationDestroy;
private Boolean parallel;
private Boolean persistent;
@@ -46,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.
@@ -65,77 +86,60 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
* @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());
Assert.state(this.asyncEventListener != null, "AsyncEventListener must not be null");
if (batchSize != null) {
asyncEventQueueFactory.setBatchSize(batchSize);
}
AsyncEventQueueFactory asyncEventQueueFactory =
this.factory != null ? (AsyncEventQueueFactory) this.factory : this.cache.createAsyncEventQueueFactory();
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 (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();
}
@@ -145,22 +149,28 @@ 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.");
"Setting an AsyncEventListener is not allowed once the AsyncEventQueue has been created");
this.asyncEventListener = listener;
}
/**
* @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;
@@ -171,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;
@@ -185,36 +196,69 @@ 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;
}
/**
* 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.
*
* @param forwardExpirationDestroy boolean value indicating whether to forward expiration destroy events.
* @see org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory#setForwardExpirationDestroy(boolean)
* @see org.apache.geode.cache.ExpirationAttributes#getAction()
* @see org.apache.geode.cache.ExpirationAction#DESTROY
*/
public void setForwardExpirationDestroy(Boolean forwardExpirationDestroy) {
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;
}
@@ -233,5 +277,4 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
public void setPersistent(Boolean persistent) {
this.persistent = persistent;
}
}

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

@@ -46,12 +46,6 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
private int remoteDistributedSystemId;
private GatewaySender gatewaySender;
private List<GatewayEventFilter> eventFilters;
private List<GatewayTransportFilter> transportFilters;
private Boolean diskSynchronous;
private Boolean batchConflationEnabled;
private Boolean parallel;
@@ -59,6 +53,10 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
private GatewayEventSubstitutionFilter eventSubstitutionFilter;
private GatewaySender gatewaySender;
private GatewaySender.OrderPolicy orderPolicy;
private Integer alertThreshold;
private Integer batchSize;
private Integer batchTimeInterval;
@@ -67,8 +65,11 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
private Integer socketBufferSize;
private Integer socketReadTimeout;
private List<GatewayEventFilter> eventFilters;
private List<GatewayTransportFilter> transportFilters;
private String diskStoreReference;
private String orderPolicy;
/**
* Constructs an instance of the {@link GatewaySenderFactoryBean} class initialized with a reference to
@@ -86,8 +87,9 @@ 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) this.factory : this.cache.createGatewaySenderFactory();
if (alertThreshold != null) {
gatewaySenderFactory.setAlertThreshold(alertThreshold);
@@ -132,12 +134,10 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
}
if (orderPolicy != null) {
Assert.isTrue(isSerialGatewaySender(), "Order Policy cannot be used with a Parallel Gateway Sender Queue.");
Assert.isTrue(VALID_ORDER_POLICIES.contains(orderPolicy.toUpperCase()),
String.format("The value for Order Policy '%s' is invalid.", orderPolicy));
Assert.isTrue(isSerialGatewaySender(), "OrderPolicy cannot be used with a Parallel GatewaySender");
gatewaySenderFactory.setOrderPolicy(GatewaySender.OrderPolicy.valueOf(orderPolicy.toUpperCase()));
gatewaySenderFactory.setOrderPolicy(this.orderPolicy);
}
gatewaySenderFactory.setParallel(isParallelGatewaySender());
@@ -162,20 +162,14 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
gatewaySender = wrapper;
}
/**
* @inheritDoc
*/
@Override
public GatewaySender getObject() throws Exception {
return gatewaySender;
return this.gatewaySender;
}
/**
* @inheritDoc
*/
@Override
public Class<?> getObjectType() {
return (gatewaySender != null ? gatewaySender.getClass() : GatewaySender.class);
return this.gatewaySender != null ? this.gatewaySender.getClass() : GatewaySender.class;
}
public void setAlertThreshold(Integer alertThreshold) {
@@ -235,6 +229,10 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
}
public void setOrderPolicy(String orderPolicy) {
setOrderPolicy(GatewaySender.OrderPolicy.valueOf(String.valueOf(orderPolicy).toUpperCase()));
}
public void setOrderPolicy(GatewaySender.OrderPolicy orderPolicy) {
this.orderPolicy = orderPolicy;
}