From 57a47844ab07c56568e26734271da14e3d819b8b Mon Sep 17 00:00:00 2001 From: John Blum Date: Wed, 25 Feb 2015 15:18:42 -0800 Subject: [PATCH] SGF-376 - The GemFire WAN GatewayHub support needs refactoring and test coverage. --- .../data/gemfire/config/GatewayHubParser.java | 142 +++++--- .../wan/AbstractWANComponentFactoryBean.java | 8 +- .../gemfire/wan/GatewayHubFactoryBean.java | 266 +++++++------- .../data/gemfire/wan/GatewayProxy.java | 181 +++++----- .../gemfire/config/spring-gemfire-1.5.xsd | 136 ++++---- .../wan/GatewayHubFactoryBeanTest.java | 325 ++++++++++++++++++ 6 files changed, 715 insertions(+), 343 deletions(-) create mode 100644 src/test/java/org/springframework/data/gemfire/wan/GatewayHubFactoryBeanTest.java diff --git a/src/main/java/org/springframework/data/gemfire/config/GatewayHubParser.java b/src/main/java/org/springframework/data/gemfire/config/GatewayHubParser.java index 99e3dc60..b98b67ce 100644 --- a/src/main/java/org/springframework/data/gemfire/config/GatewayHubParser.java +++ b/src/main/java/org/springframework/data/gemfire/config/GatewayHubParser.java @@ -17,6 +17,7 @@ package org.springframework.data.gemfire.config; import java.util.List; +import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.ManagedList; import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; @@ -24,89 +25,126 @@ import org.springframework.beans.factory.xml.ParserContext; import org.springframework.data.gemfire.wan.GatewayHubFactoryBean; import org.springframework.data.gemfire.wan.GatewayProxy; import org.springframework.util.CollectionUtils; -import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; import org.w3c.dom.Element; /** + * Parser for the <gateway-hub> SDG XML namespace element used to create GemFire GatewayHubs. + * * @author David Turanski - * + * @author John J. Blum + * @see org.springframework.beans.factory.config.BeanDefinition + * @see org.springframework.beans.factory.support.BeanDefinitionBuilder + * @see org.springframework.beans.factory.support.ManagedList + * @see org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser + * @see org.springframework.beans.factory.xml.ParserContext + * @see org.springframework.data.gemfire.wan.GatewayHubFactoryBean + * @see org.springframework.data.gemfire.wan.GatewayProxy */ class GatewayHubParser extends AbstractSimpleBeanDefinitionParser { + @Override protected Class getBeanClass(Element element) { return GatewayHubFactoryBean.class; } - @SuppressWarnings({ "rawtypes", "unchecked" }) @Override + @SuppressWarnings({ "rawtypes", "unchecked" }) protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + builder.addConstructorArgReference(ParsingUtils.resolveCacheReference(element.getAttribute("cache-ref"))); builder.setLazyInit(false); - String cacheRef = element.getAttribute("cache-ref"); - // add cache reference (fallback to default if nothing is specified) - builder.addConstructorArgReference((StringUtils.hasText(cacheRef) ? cacheRef : "gemfireCache")); + ParsingUtils.setPropertyValue(element, builder, "bind-address"); ParsingUtils.setPropertyValue(element, builder, "manual-start"); + //ParsingUtils.setPropertyValue(element, builder, "max-time-between-pings", "maximumTimeBetweenPings"); ParsingUtils.setPropertyValue(element, builder, "socket-buffer-size"); ParsingUtils.setPropertyValue(element, builder, "startup-policy"); ParsingUtils.setPropertyValue(element, builder, "port"); - + + parseGateways(element, parserContext, builder); + } + + private void parseGateways(Element element, ParserContext parserContext, BeanDefinitionBuilder gatewayHubBuilder) { List gatewayElements = DomUtils.getChildElementsByTagName(element, "gateway"); + if (!CollectionUtils.isEmpty(gatewayElements)) { - ManagedList gateways = new ManagedList(); + ManagedList gateways = new ManagedList(); + for (Element gatewayElement : gatewayElements) { BeanDefinitionBuilder gatewayBuilder = BeanDefinitionBuilder.genericBeanDefinition(GatewayProxy.class); + ParsingUtils.setPropertyValue(gatewayElement, gatewayBuilder, "gateway-id", "id"); ParsingUtils.setPropertyValue(gatewayElement, gatewayBuilder, "concurrency-level"); - ParsingUtils.setPropertyValue(gatewayElement, gatewayBuilder, "socket-read-timeout"); - ParsingUtils.setPropertyValue(gatewayElement, gatewayBuilder, "socket-buffer-size"); ParsingUtils.setPropertyValue(gatewayElement, gatewayBuilder, "order-policy"); - List endpointElements = DomUtils.getChildElementsByTagName(gatewayElement, "gateway-endpoint"); - if (!CollectionUtils.isEmpty(endpointElements)) { - ManagedList endpoints = new ManagedList(); - for (Element endpointElement : endpointElements) { - BeanDefinitionBuilder endpointBuilder = BeanDefinitionBuilder - .genericBeanDefinition(GatewayProxy.GatewayEndpoint.class); - ParsingUtils.setPropertyValue(endpointElement, endpointBuilder, "host"); - ParsingUtils.setPropertyValue(endpointElement, endpointBuilder, "port"); - ParsingUtils.setPropertyValue(endpointElement, endpointBuilder, "endpoint-id", "id"); - endpoints.add(endpointBuilder.getBeanDefinition()); - } - gatewayBuilder.addPropertyValue("endpoints", endpoints); - } + ParsingUtils.setPropertyValue(gatewayElement, gatewayBuilder, "socket-buffer-size"); + ParsingUtils.setPropertyValue(gatewayElement, gatewayBuilder, "socket-read-timeout"); - Element gatewayListenerElement = DomUtils.getChildElementByTagName(gatewayElement, "gateway-listener"); - if (gatewayListenerElement != null) { - Object obj = ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, gatewayListenerElement, - gatewayBuilder); - gatewayBuilder.addPropertyValue("listeners", obj); - } + parseGatewayEndpoints(gatewayElement, gatewayBuilder); + parseGatewayListener(gatewayElement, parserContext, gatewayBuilder); + + parseGatewayQueue(gatewayElement, gatewayBuilder); - Element gatewayQueueElement = DomUtils.getChildElementByTagName(gatewayElement, "gateway-queue"); - if (gatewayQueueElement != null) { - BeanDefinitionBuilder queueBuilder = BeanDefinitionBuilder - .genericBeanDefinition(GatewayProxy.GatewayQueue.class); - ParsingUtils.setPropertyValue(gatewayQueueElement, queueBuilder, "alert-threshold"); - ParsingUtils.setPropertyValue(gatewayQueueElement, queueBuilder, "batch-size"); - ParsingUtils.setPropertyValue(gatewayQueueElement, queueBuilder, "batch-time-interval"); - ParsingUtils.setPropertyValue(gatewayQueueElement, queueBuilder, "maximum-queue-memory"); - ParsingUtils.setPropertyValue(gatewayQueueElement, queueBuilder, "persistent"); - ParsingUtils.setPropertyValue(gatewayQueueElement, queueBuilder, "disk-store-ref"); - /* - * Make sure any disk store is created first - */ - if (gatewayQueueElement.hasAttribute("disk-store-ref")) { - gatewayBuilder.getBeanDefinition().setDependsOn( - new String[] {gatewayQueueElement.getAttribute("disk-store-ref")}); - } - ParsingUtils.setPropertyValue(gatewayQueueElement, queueBuilder, "enable-batch-conflation"); - gatewayBuilder.addPropertyValue("queue", queueBuilder.getBeanDefinition()); - - - } gateways.add(gatewayBuilder.getBeanDefinition()); } - builder.addPropertyValue("gateways", gateways); + + gatewayHubBuilder.addPropertyValue("gateways", gateways); } } + + private void parseGatewayEndpoints(Element gatewayElement, BeanDefinitionBuilder gatewayBuilder) { + List endpointElements = DomUtils.getChildElementsByTagName(gatewayElement, "gateway-endpoint"); + + if (!CollectionUtils.isEmpty(endpointElements)) { + ManagedList endpoints = new ManagedList(); + + for (Element endpointElement : endpointElements) { + BeanDefinitionBuilder endpointBuilder = BeanDefinitionBuilder.genericBeanDefinition( + GatewayProxy.GatewayEndpoint.class); + + ParsingUtils.setPropertyValue(endpointElement, endpointBuilder, "endpoint-id", "id"); + ParsingUtils.setPropertyValue(endpointElement, endpointBuilder, "host"); + ParsingUtils.setPropertyValue(endpointElement, endpointBuilder, "port"); + + endpoints.add(endpointBuilder.getBeanDefinition()); + } + + gatewayBuilder.addPropertyValue("endpoints", endpoints); + } + } + + private void parseGatewayListener(Element gatewayElement, ParserContext parserContext, + BeanDefinitionBuilder gatewayBuilder) { + + Element gatewayListenerElement = DomUtils.getChildElementByTagName(gatewayElement, "gateway-listener"); + + if (gatewayListenerElement != null) { + gatewayBuilder.addPropertyValue("listeners", ParsingUtils.parseRefOrNestedBeanDeclaration( + parserContext, gatewayListenerElement, gatewayBuilder)); + } + } + + private void parseGatewayQueue(Element gatewayElement, BeanDefinitionBuilder gatewayBuilder) { + Element queueElement = DomUtils.getChildElementByTagName(gatewayElement, "gateway-queue"); + + if (queueElement != null) { + BeanDefinitionBuilder queueBuilder = BeanDefinitionBuilder.genericBeanDefinition( + GatewayProxy.GatewayQueue.class); + + ParsingUtils.setPropertyValue(queueElement, queueBuilder, "alert-threshold"); + ParsingUtils.setPropertyValue(queueElement, queueBuilder, "batch-size"); + ParsingUtils.setPropertyValue(queueElement, queueBuilder, "batch-time-interval"); + ParsingUtils.setPropertyValue(queueElement, queueBuilder, "disk-store-ref"); + ParsingUtils.setPropertyValue(queueElement, queueBuilder, "enable-batch-conflation"); + ParsingUtils.setPropertyValue(queueElement, queueBuilder, "maximum-queue-memory"); + ParsingUtils.setPropertyValue(queueElement, queueBuilder, "persistent"); + + if (queueElement.hasAttribute("disk-store-ref")) { + gatewayBuilder.getBeanDefinition().setDependsOn(new String[] { + queueElement.getAttribute("disk-store-ref") }); + } + + gatewayBuilder.addPropertyValue("queue", queueBuilder.getBeanDefinition()); + } + } + } diff --git a/src/main/java/org/springframework/data/gemfire/wan/AbstractWANComponentFactoryBean.java b/src/main/java/org/springframework/data/gemfire/wan/AbstractWANComponentFactoryBean.java index b701aebb..4a54cf77 100644 --- a/src/main/java/org/springframework/data/gemfire/wan/AbstractWANComponentFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/wan/AbstractWANComponentFactoryBean.java @@ -35,8 +35,8 @@ import com.gemstone.gemfire.management.internal.cli.util.spring.StringUtils; * @author David Turanski * @author John Blum */ -public abstract class AbstractWANComponentFactoryBean implements BeanNameAware, FactoryBean, InitializingBean, - DisposableBean { +public abstract class AbstractWANComponentFactoryBean implements BeanNameAware, FactoryBean, + InitializingBean, DisposableBean { protected static final List VALID_ORDER_POLICIES = Arrays.asList("KEY", "PARTITION", "THREAD"); @@ -67,7 +67,7 @@ public abstract class AbstractWANComponentFactoryBean implements BeanNameAwar } public String getName() { - return (StringUtils.hasText(name) ? name: beanName); + return (StringUtils.hasText(name) ? name : beanName); } @Override @@ -83,8 +83,8 @@ public abstract class AbstractWANComponentFactoryBean implements BeanNameAwar @Override public final void afterPropertiesSet() throws Exception { - Assert.notNull(getName(), "Name must not be null."); Assert.notNull(cache, "Cache must not be null."); + Assert.notNull(getName(), "Name must not be null."); doInit(); } diff --git a/src/main/java/org/springframework/data/gemfire/wan/GatewayHubFactoryBean.java b/src/main/java/org/springframework/data/gemfire/wan/GatewayHubFactoryBean.java index 1ab35dd8..67ca34e5 100644 --- a/src/main/java/org/springframework/data/gemfire/wan/GatewayHubFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/wan/GatewayHubFactoryBean.java @@ -17,11 +17,11 @@ package org.springframework.data.gemfire.wan; import java.io.IOException; import java.util.Arrays; +import java.util.Collections; import java.util.List; import org.springframework.data.gemfire.wan.GatewayProxy.GatewayQueue; import org.springframework.util.Assert; -import org.springframework.util.CollectionUtils; import com.gemstone.gemfire.cache.Cache; import com.gemstone.gemfire.cache.util.Gateway; @@ -29,36 +29,43 @@ import com.gemstone.gemfire.cache.util.Gateway.OrderPolicy; import com.gemstone.gemfire.cache.util.GatewayEventListener; import com.gemstone.gemfire.cache.util.GatewayHub; import com.gemstone.gemfire.cache.util.GatewayQueueAttributes; +import com.gemstone.gemfire.management.internal.cli.util.spring.StringUtils; /** - * FactoryBean for creating a GemFire {@link GatewayHub} (deprecated in Gemfire - * 7) + * FactoryBean for creating a GemFire {@link GatewayHub} (deprecated in Gemfire 7). + * * @author David Turanski - * + * @author John Blum + * @see org.springframework.data.gemfire.wan.AbstractWANComponentFactoryBean + * @see com.gemstone.gemfire.cache.Cache + * @see com.gemstone.gemfire.cache.util.Gateway + * @see com.gemstone.gemfire.cache.util.GatewayHub + * @see com.gemstone.gemfire.cache.util.GatewayEventListener + * @see com.gemstone.gemfire.cache.util.GatewayQueueAttributes */ +@SuppressWarnings({ "deprecation", "unused" }) public class GatewayHubFactoryBean extends AbstractWANComponentFactoryBean { - private static List validStartupPolicyValues = Arrays.asList("none", "primary", "secondary"); - private static List validOrderPolicyValues = Arrays.asList("KEY,PARTITION,THREAD"); + private static List validOrderPolicyValues = Arrays.asList("key, partition, thread"); - private GatewayHub gatewayHub; - - private Integer port; - - private String bindAddress; - - private Integer maximumTimeBetweenPings; - - private Integer socketBufferSize; - - private String startupPolicy; + private static List validStartupPolicyValues = Arrays.asList(GatewayHub.STARTUP_POLICY_NONE, + GatewayHub.STARTUP_POLICY_PRIMARY, GatewayHub.STARTUP_POLICY_SECONDARY); private Boolean manualStart; + private GatewayHub gatewayHub; + + private Integer maximumTimeBetweenPings; + private Integer port; + private Integer socketBufferSize; + private List gateways; + private String bindAddress; + private String startupPolicy; + /** - * @param cache the Gemfire cache + * @param cache a reference to the GemFire Cache. */ public GatewayHubFactoryBean(Cache cache) { super(cache); @@ -71,134 +78,143 @@ public class GatewayHubFactoryBean extends AbstractWANComponentFactoryBean getObjectType() { - return GatewayHub.class; + return (gatewayHub != null ? gatewayHub.getClass() : GatewayHub.class); } @Override protected void doInit() { - String name = getName(); - gatewayHub = cache.addGatewayHub(name, port == null ? GatewayHub.DEFAULT_PORT : port); + gatewayHub = cache.addGatewayHub(getName(), getPort()); if (log.isDebugEnabled()) { - log.debug("added gateway hub " + name); + log.debug(String.format("Adding GemFire GatewayHub (%1$s)", getName())); } - Assert.notNull(cache.getGatewayHub(name)); + Assert.notNull(cache.getGatewayHub(getName())); - if (bindAddress != null) { - gatewayHub.setBindAddress(bindAddress); - } - if (manualStart != null) { - gatewayHub.setManualStart(manualStart); - } - if (socketBufferSize != null) { - gatewayHub.setSocketBufferSize(socketBufferSize); - } - if (startupPolicy != null) { - Assert.isTrue(validStartupPolicyValues.contains(startupPolicy), "The value of startup policy:'" - + startupPolicy + "' is invalid"); - gatewayHub.setStartupPolicy(startupPolicy); - } - if (maximumTimeBetweenPings != null) { - gatewayHub.setMaximumTimeBetweenPings(maximumTimeBetweenPings); - } - - if (!CollectionUtils.isEmpty(gateways)) { - configureGateways(); - } - - if (gatewayHub.getManualStart() == false) { - try { - gatewayHub.start(); - } catch (IOException e) { - throw new RuntimeException(e); - } - } + gatewayHub.setBindAddress(getBindAddress()); + gatewayHub.setManualStart(isManualStart(GatewayHub.DEFAULT_MANUAL_START)); + gatewayHub.setMaximumTimeBetweenPings(getMaximumTimeBetweenPings()); + gatewayHub.setSocketBufferSize(getSocketBufferSize()); + + String localStartupPolicy = getStartupPolicy().trim().toLowerCase(); + + Assert.isTrue(validStartupPolicyValues.contains(localStartupPolicy), String.format( + "The specified startup-policy '%1$s' is not valid!", localStartupPolicy)); + + gatewayHub.setStartupPolicy(localStartupPolicy); + + configureGateways(); + autoStart(); } - public void setPort(Integer port) { - this.port = port; - } - - public void setBindAddress(String bindAddress) { - this.bindAddress = bindAddress; - } - - public void setMaximumTimeBetweenPings(Integer maximumTimeBetweenPings) { - this.maximumTimeBetweenPings = maximumTimeBetweenPings; - } - - public void setSocketBufferSize(Integer socketBufferSize) { - this.socketBufferSize = socketBufferSize; - } - - public void setStartupPolicy(String startupPolicy) { - this.startupPolicy = startupPolicy; - } - - public void setManualStart(Boolean manualStart) { - this.manualStart = manualStart; - } - - public void setGateways(List gateways) { - this.gateways = gateways; - } - private void configureGateways() { - for (GatewayProxy gateway : gateways) { - Gateway gw = gatewayHub.addGateway( - gateway.getId(), - gateway.getConcurrencyLevel() == null ? Gateway.DEFAULT_CONCURRENCY_LEVEL : gateway - .getConcurrencyLevel()); - if (!CollectionUtils.isEmpty(gateway.getEndpoints())) { - for (GatewayProxy.GatewayEndpoint endpoint : gateway.getEndpoints()) { - gw.addEndpoint(endpoint.getId(), endpoint.getHost(), endpoint.getPort()); - } - } - if (!CollectionUtils.isEmpty(gateway.getListeners())) { - for (GatewayEventListener listener : gateway.getListeners()) { - gw.addListener(listener); - } - } - if (gateway.getOrderPolicy() != null) { - Assert.isTrue(validOrderPolicyValues.contains(gateway.getOrderPolicy()), - "The value of order policy:'" + gateway.getOrderPolicy() + "' is invalid"); - gw.setOrderPolicy(OrderPolicy.valueOf(gateway.getOrderPolicy())); - } - if (gateway.getSocketBufferSize() != null) { - gw.setSocketBufferSize(gateway.getSocketBufferSize()); + for (GatewayProxy gatewayProxy : getGateways()) { + Gateway gateway = gatewayHub.addGateway(gatewayProxy.getId(), gatewayProxy.getConcurrencyLevel()); + + for (GatewayProxy.GatewayEndpoint endpoint : gatewayProxy.getEndpoints()) { + gateway.addEndpoint(endpoint.getId(), endpoint.getHost(), endpoint.getPort()); } - if (gateway.getQueue() != null) { - GatewayQueue queue = gateway.getQueue(); - GatewayQueueAttributes queueAttributes = gw.getQueueAttributes(); - if (queue.getAlertThreshold() != null) { - queueAttributes.setAlertThreshold(queue.getAlertThreshold()); - } - if (queue.getEnableBatchConflation() != null) { - queueAttributes.setBatchConflation(queue.getEnableBatchConflation()); - } - if (queue.getBatchSize() != null) { - queueAttributes.setBatchSize(queue.getBatchSize()); - } - if (queue.getBatchTimeInterval() != null) { - queueAttributes.setBatchTimeInterval(queue.getBatchTimeInterval()); - } + for (GatewayEventListener listener : gatewayProxy.getListeners()) { + gateway.addListener(listener); + } + + if (StringUtils.hasText(gatewayProxy.getOrderPolicy())) { + String orderPolicyValue = gatewayProxy.getOrderPolicy().trim().toUpperCase(); + OrderPolicy orderPolicy = OrderPolicy.valueOf(orderPolicyValue); + Assert.notNull(orderPolicy, String.format("The specified order-policy '%1$s' is not valid!", + orderPolicyValue)); + gateway.setOrderPolicy(orderPolicy); + } + + gateway.setSocketBufferSize(gatewayProxy.getSocketBufferSize()); + + if (gatewayProxy.getQueue() != null) { + GatewayQueue queue = gatewayProxy.getQueue(); + GatewayQueueAttributes queueAttributes = gateway.getQueueAttributes(); + + queueAttributes.setAlertThreshold(queue.getAlertThreshold()); + queueAttributes.setBatchConflation(queue.getEnableBatchConflation()); + queueAttributes.setBatchSize(queue.getBatchSize()); + queueAttributes.setBatchTimeInterval(queue.getBatchTimeInterval()); + queueAttributes.setEnablePersistence(queue.getPersistent()); + queueAttributes.setMaximumQueueMemory(queue.getMaximumQueueMemory()); if (queue.getDiskStoreRef() != null) { boolean persistent = (queue.getPersistent() == null) ? Boolean.TRUE : queue.getPersistent(); Assert.isTrue(persistent, "specifying a disk store requires persistent property to be true"); queueAttributes.setDiskStoreName(queue.getDiskStoreRef()); } - - if (queue.getPersistent() != null) { - queueAttributes.setEnablePersistence(queue.getPersistent()); - } - - if (queue.getMaximumQueueMemory() != null) { - queueAttributes.setMaximumQueueMemory(queue.getMaximumQueueMemory()); - } } } } + + private void autoStart() { + if (!gatewayHub.getManualStart()) { + try { + gatewayHub.start(); + } + catch (IOException e) { + throw new RuntimeException(e); + } + } + } + + public void setBindAddress(String bindAddress) { + this.bindAddress = bindAddress; + } + + String getBindAddress() { + return (StringUtils.hasText(bindAddress) ? bindAddress : GatewayHub.DEFAULT_BIND_ADDRESS); + } + + public void setGateways(List gateways) { + this.gateways = gateways; + } + + List getGateways() { + return (gateways != null ? gateways : Collections.emptyList()); + } + + public void setManualStart(Boolean manualStart) { + this.manualStart = manualStart; + } + + boolean isManualStart(final boolean defaultManualStart) { + return (manualStart != null ? manualStart : defaultManualStart); + } + + public void setMaximumTimeBetweenPings(Integer maximumTimeBetweenPings) { + this.maximumTimeBetweenPings = maximumTimeBetweenPings; + } + + Integer getMaximumTimeBetweenPings() { + return (maximumTimeBetweenPings != null ? maximumTimeBetweenPings + : GatewayHub.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS); + } + + public void setPort(Integer port) { + this.port = port; + } + + Integer getPort() { + return (port != null ? port : GatewayHub.DEFAULT_PORT); + } + + public void setSocketBufferSize(Integer socketBufferSize) { + this.socketBufferSize = socketBufferSize; + } + + Integer getSocketBufferSize() { + return (socketBufferSize != null ? socketBufferSize : GatewayHub.DEFAULT_SOCKET_BUFFER_SIZE); + } + + public void setStartupPolicy(String startupPolicy) { + this.startupPolicy = startupPolicy; + } + + String getStartupPolicy() { + return (StringUtils.hasText(startupPolicy) ? startupPolicy : GatewayHub.DEFAULT_STARTUP_POLICY); + } + } diff --git a/src/main/java/org/springframework/data/gemfire/wan/GatewayProxy.java b/src/main/java/org/springframework/data/gemfire/wan/GatewayProxy.java index 62e81af2..100230c8 100644 --- a/src/main/java/org/springframework/data/gemfire/wan/GatewayProxy.java +++ b/src/main/java/org/springframework/data/gemfire/wan/GatewayProxy.java @@ -15,94 +15,108 @@ */ package org.springframework.data.gemfire.wan; +import java.util.Collections; import java.util.List; import com.gemstone.gemfire.cache.util.Gateway; import com.gemstone.gemfire.cache.util.GatewayEventListener; +import com.gemstone.gemfire.cache.util.GatewayQueueAttributes; /** - * This class used to allow decoupling of 'gateway' parsing from 'gateway-hub' - * parsing + * This class used to allow decoupling of 'gateway' parsing from 'gateway-hub' parsing. * * @author David Turanski - * + * @author John Blum + * @see com.gemstone.gemfire.cache.util.Gateway */ +@SuppressWarnings({ "deprecation", "unused" }) public class GatewayProxy { + private GatewayQueue queue; + + private Integer concurrencyLevel; + private Integer socketBufferSize; + //private Integer socketReadTimeout; + private List endpoints; - - private Integer concurrencyLevel = Gateway.DEFAULT_CONCURRENCY_LEVEL; - - private String id; - private List listeners; + private String id; private String orderPolicy; - private int socketBufferSize = Gateway.DEFAULT_SOCKET_BUFFER_SIZE; + public void setConcurrencyLevel(Integer concurrencyLevel) { + this.concurrencyLevel = concurrencyLevel; + } - private GatewayQueue queue; + public Integer getConcurrencyLevel() { + return (concurrencyLevel != null ? concurrencyLevel : Gateway.DEFAULT_CONCURRENCY_LEVEL); + } public void setEndpoints(List endpoints) { this.endpoints = endpoints; } + public List getEndpoints() { + return (endpoints != null ? endpoints : Collections.emptyList()); + } + + public void setId(String id) { + this.id = id; + } + + public String getId() { + return id; + } + public void setListeners(List listeners) { this.listeners = listeners; } + public List getListeners() { + return (listeners != null ? listeners : Collections.emptyList()); + } + + public void setOrderPolicy(String orderPolicy) { + this.orderPolicy = orderPolicy; + } + + public String getOrderPolicy() { + return orderPolicy; + } + public void setQueue(GatewayQueue queue) { this.queue = queue; } public GatewayQueue getQueue() { - return this.queue; - } - - public Integer getConcurrencyLevel() { - return this.concurrencyLevel; - } - - public List getEndpoints() { - return endpoints; - } - - public String getId() { - return this.id; - } - - public List getListeners() { - return this.listeners; - } - - public String getOrderPolicy() { - return this.orderPolicy; - } - - public Integer getSocketBufferSize() { - return this.socketBufferSize; - } - - public void setId(String id) { - this.id = id; - } - - public void setOrderPolicy(String orderPolicy) { - this.orderPolicy = orderPolicy; + return queue; } public void setSocketBufferSize(int socketBufferSize) { this.socketBufferSize = socketBufferSize; - } - public static class GatewayEndpoint { - private String host; + public Integer getSocketBufferSize() { + return (socketBufferSize != null ? socketBufferSize : Gateway.DEFAULT_SOCKET_BUFFER_SIZE); + } - private String id; + /* + public void setSocketReadTimeout(final Integer socketReadTimeout) { + this.socketReadTimeout = socketReadTimeout; + } + + public Integer getSocketReadTimeout() { + return (socketReadTimeout != null ? socketReadTimeout : Gateway.DEFAULT_SOCKET_READ_TIMEOUT); + } + */ + + public static class GatewayEndpoint { private int port; + private String id; + private String host; + public String getHost() { return host; } @@ -129,75 +143,74 @@ public class GatewayProxy { } public static class GatewayQueue { - private Integer alertThreshold; private Boolean enableBatchConflation; - - private Integer batchTimeInterval; - - private Integer batchSize; - private Boolean persistent; - private String diskStoreRef; - + private Integer alertThreshold; + private Integer batchSize; + private Integer batchTimeInterval; private Integer maximumQueueMemory; - public Integer getAlertThreshold() { - return alertThreshold; - } + private String diskStoreRef; public void setAlertThreshold(Integer alertThreshold) { this.alertThreshold = alertThreshold; } - public Boolean getEnableBatchConflation() { - return enableBatchConflation; - } - - public void setEnableBatchConflation(Boolean enableBatchConflation) { - this.enableBatchConflation = enableBatchConflation; - } - - public Integer getBatchTimeInterval() { - return batchTimeInterval; - } - - public void setBatchTimeInterval(Integer batchTimeInterval) { - this.batchTimeInterval = batchTimeInterval; - } - - public Integer getBatchSize() { - return batchSize; + public Integer getAlertThreshold() { + return (alertThreshold != null ? alertThreshold : GatewayQueueAttributes.DEFAULT_ALERT_THRESHOLD); } public void setBatchSize(Integer batchSize) { this.batchSize = batchSize; } - public Boolean getPersistent() { - return persistent; + public Integer getBatchSize() { + return (batchSize != null ? batchSize : GatewayQueueAttributes.DEFAULT_BATCH_SIZE); } - public void setPersistent(Boolean persistent) { - this.persistent = persistent; + public void setBatchTimeInterval(Integer batchTimeInterval) { + this.batchTimeInterval = batchTimeInterval; } - public String getDiskStoreRef() { - return diskStoreRef; + public Integer getBatchTimeInterval() { + return (batchTimeInterval != null ? batchTimeInterval : GatewayQueueAttributes.DEFAULT_BATCH_TIME_INTERVAL); } public void setDiskStoreRef(String diskStoreRef) { this.diskStoreRef = diskStoreRef; } - public Integer getMaximumQueueMemory() { - return maximumQueueMemory; + public String getDiskStoreRef() { + return diskStoreRef; + } + + public void setEnableBatchConflation(Boolean enableBatchConflation) { + this.enableBatchConflation = enableBatchConflation; + } + + public Boolean getEnableBatchConflation() { + return (enableBatchConflation != null ? enableBatchConflation + : GatewayQueueAttributes.DEFAULT_BATCH_CONFLATION); } public void setMaximumQueueMemory(Integer maximumQueueMemory) { this.maximumQueueMemory = maximumQueueMemory; } + public Integer getMaximumQueueMemory() { + return (maximumQueueMemory != null ? maximumQueueMemory + : GatewayQueueAttributes.DEFAULT_MAXIMUM_QUEUE_MEMORY); + } + + public void setPersistent(Boolean persistent) { + this.persistent = persistent; + } + + public Boolean getPersistent() { + return (persistent != null ? persistent : GatewayQueueAttributes.DEFAULT_ENABLE_PERSISTENCE); + } } + } diff --git a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.5.xsd b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.5.xsd index 58ae0818..6bf942d7 100644 --- a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.5.xsd +++ b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.5.xsd @@ -2894,7 +2894,7 @@ The id of the cache - default is gemfireCache - + @@ -2957,7 +2957,7 @@ Scope also determines whether to allow remote invocation of some of the region - + - + - + - - - - - - + - - - - - - + - + + + + + + + + + + + - - - - - - + + + @@ -3081,24 +3073,7 @@ Deprecated as of Gemfire 7 Deprecated as of Gemfire 7 ]]> - - - - - - - - - - - + - + - + - + + + + + + + + + + + @@ -3146,7 +3133,7 @@ Deprecated as of Gemfire 7 - + - + - + @@ -3191,24 +3176,14 @@ Specifies the id for this gateway ]]> - + - - - - - - + - + + + + + + @@ -3236,7 +3217,6 @@ Deprecated as of Gemfire 7 - diff --git a/src/test/java/org/springframework/data/gemfire/wan/GatewayHubFactoryBeanTest.java b/src/test/java/org/springframework/data/gemfire/wan/GatewayHubFactoryBeanTest.java new file mode 100644 index 00000000..3bdc136f --- /dev/null +++ b/src/test/java/org/springframework/data/gemfire/wan/GatewayHubFactoryBeanTest.java @@ -0,0 +1,325 @@ +/* + * Copyright 2010-2013 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.wan; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.eq; +import static org.mockito.Matchers.same; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.Arrays; +import java.util.List; + +import org.junit.Before; +import org.junit.Test; + +import com.gemstone.gemfire.cache.Cache; +import com.gemstone.gemfire.cache.util.Gateway; +import com.gemstone.gemfire.cache.util.GatewayEventListener; +import com.gemstone.gemfire.cache.util.GatewayHub; +import com.gemstone.gemfire.cache.util.GatewayQueueAttributes; + +/** + * The GatewayHubFactoryBeanTest class is a test suite of test cases testing the contract and functionality + * of the GatewayHubFactoryBean. + * + * @author John Blum + * @see org.junit.Test + * @see org.mockito.Mockito + * @see org.springframework.data.gemfire.wan.GatewayHubFactoryBean + * @since 1.5.3 + */ +@SuppressWarnings("deprecation") +public class GatewayHubFactoryBeanTest { + + private Cache mockCache; + + private GatewayHubFactoryBean factoryBean; + + @Before + public void setup() { + mockCache = mock(Cache.class, "GemFire Cache"); + factoryBean = new GatewayHubFactoryBean(mockCache); + } + + @Test + public void testGetObjectAndObjectType() throws Exception { + assertNull(factoryBean.getObject()); + assertEquals(GatewayHub.class, factoryBean.getObjectType()); + } + + @Test + public void testSetAndGetBindAddress() { + assertEquals(GatewayHub.DEFAULT_BIND_ADDRESS, factoryBean.getBindAddress()); + factoryBean.setBindAddress("10.127.255.1"); + assertEquals("10.127.255.1", factoryBean.getBindAddress()); + factoryBean.setBindAddress(null); + assertEquals(GatewayHub.DEFAULT_BIND_ADDRESS, factoryBean.getBindAddress()); + } + + @Test + public void testGetGateways() { + List gateways = factoryBean.getGateways(); + + assertNotNull(gateways); + assertTrue(gateways.isEmpty()); + } + + @Test + public void testSetAndIsManualStart() { + assertEquals(GatewayHub.DEFAULT_MANUAL_START, factoryBean.isManualStart(GatewayHub.DEFAULT_MANUAL_START)); + factoryBean.setManualStart(true); + assertTrue(factoryBean.isManualStart(GatewayHub.DEFAULT_MANUAL_START)); + factoryBean.setManualStart(false); + assertFalse(factoryBean.isManualStart(true)); + factoryBean.setManualStart(null); + assertEquals(GatewayHub.DEFAULT_MANUAL_START, factoryBean.isManualStart(GatewayHub.DEFAULT_MANUAL_START)); + } + + /* + @Test + public void testSetAndGetMaxConnections() { + assertEquals(GatewayHub.DEFAULT_MAX_CONNECTIONS, factoryBean.getMaxConnections().intValue()); + factoryBean.setMaxConnections(8192); + assertEquals(8192, factoryBean.getMaxConnections().intValue()); + factoryBean.setMaxConnections(null); + assertEquals(GatewayHub.DEFAULT_MAX_CONNECTIONS, factoryBean.getMaxConnections().intValue()); + } + */ + + @Test + public void testSetAndGetMaximumTimeBetweenPings() { + assertEquals(GatewayHub.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, factoryBean.getMaximumTimeBetweenPings().intValue()); + factoryBean.setMaximumTimeBetweenPings(15000); + assertEquals(15000, factoryBean.getMaximumTimeBetweenPings().intValue()); + factoryBean.setMaximumTimeBetweenPings(null); + assertEquals(GatewayHub.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, factoryBean.getMaximumTimeBetweenPings().intValue()); + } + + @Test + public void testSetAndGetPort() { + assertEquals(GatewayHub.DEFAULT_PORT, factoryBean.getPort().intValue()); + factoryBean.setPort(15221); + assertEquals(15221, factoryBean.getPort().intValue()); + factoryBean.setPort(null); + assertEquals(GatewayHub.DEFAULT_PORT, factoryBean.getPort().intValue()); + } + + @Test + public void testSetAndGetSocketBufferSize() { + assertEquals(GatewayHub.DEFAULT_SOCKET_BUFFER_SIZE, factoryBean.getSocketBufferSize().intValue()); + factoryBean.setSocketBufferSize(16384); + assertEquals(16384, factoryBean.getSocketBufferSize().intValue()); + factoryBean.setSocketBufferSize(null); + assertEquals(GatewayHub.DEFAULT_SOCKET_BUFFER_SIZE, factoryBean.getSocketBufferSize().intValue()); + } + + @Test + public void testSetAndGetStartUpPolicy() { + assertEquals(GatewayHub.DEFAULT_STARTUP_POLICY, factoryBean.getStartupPolicy()); + factoryBean.setStartupPolicy(GatewayHub.STARTUP_POLICY_PRIMARY); + assertEquals(GatewayHub.STARTUP_POLICY_PRIMARY, factoryBean.getStartupPolicy()); + factoryBean.setStartupPolicy(null); + assertEquals(GatewayHub.DEFAULT_STARTUP_POLICY, factoryBean.getStartupPolicy()); + factoryBean.setStartupPolicy(GatewayHub.STARTUP_POLICY_SECONDARY); + assertEquals(GatewayHub.STARTUP_POLICY_SECONDARY, factoryBean.getStartupPolicy()); + } + + @Test(expected = IllegalArgumentException.class) + public void testAfterPropertiesSetWitNullCache() throws Exception { + try { + new GatewayHubFactoryBean(null).afterPropertiesSet(); + } + catch (IllegalArgumentException expected) { + assertEquals("Cache must not be null.", expected.getMessage()); + throw expected; + } + } + + @Test(expected = IllegalArgumentException.class) + public void testAfterPropertiesSetWithNullName() throws Exception { + try { + factoryBean.afterPropertiesSet(); + } + catch (IllegalArgumentException expected) { + assertEquals("Name must not be null.", expected.getMessage()); + throw expected; + } + } + + @Test + public void testDoInit() throws Exception { + String gatewayHubName = "testDoInit"; + + GatewayProxy.GatewayEndpoint gatewayEndpointOne = new GatewayProxy.GatewayEndpoint(); + + gatewayEndpointOne.setHost("localhost"); + gatewayEndpointOne.setId("123"); + gatewayEndpointOne.setPort(2121); + + GatewayProxy.GatewayEndpoint gatewayEndpointTwo = new GatewayProxy.GatewayEndpoint(); + + gatewayEndpointOne.setHost("localhost"); + gatewayEndpointOne.setId("456"); + gatewayEndpointOne.setPort(4242); + + GatewayEventListener mockGatewayListener = mock(GatewayEventListener.class, + "testDoInit.MockGatewayEventListener"); + + GatewayProxy.GatewayQueue gatewayQueue = new GatewayProxy.GatewayQueue(); + + gatewayQueue.setAlertThreshold(20); + gatewayQueue.setBatchSize(100); + gatewayQueue.setBatchTimeInterval(60000); + gatewayQueue.setDiskStoreRef("diskX"); + gatewayQueue.setEnableBatchConflation(true); + gatewayQueue.setMaximumQueueMemory(1024); + gatewayQueue.setPersistent(true); + + GatewayProxy gatewayProxy = new GatewayProxy(); + + gatewayProxy.setId("gatewayProxyId"); + gatewayProxy.setConcurrencyLevel(4); + gatewayProxy.setEndpoints(Arrays.asList(gatewayEndpointOne, gatewayEndpointTwo)); + gatewayProxy.setListeners(Arrays.asList(mockGatewayListener)); + gatewayProxy.setOrderPolicy(" thReAD "); + gatewayProxy.setQueue(gatewayQueue); + gatewayProxy.setSocketBufferSize(16384); + //gatewayProxy.setSocketReadTimeout(300); + + GatewayHub mockGatewayHub = mock(GatewayHub.class, "testDoInit.MockGatewayHub"); + + Gateway mockGateway = mock(Gateway.class, "testDoInit.MockGateway"); + + GatewayQueueAttributes mockGatewayQueueAttributes = mock(GatewayQueueAttributes.class, + "testDoInit.MockGatewayQueueAttributes"); + + when(mockCache.addGatewayHub(eq(gatewayHubName), eq(8484))).thenReturn(mockGatewayHub); + when(mockCache.getGatewayHub(eq(gatewayHubName))).thenReturn(mockGatewayHub); + when(mockGatewayHub.addGateway(eq(gatewayProxy.getId()), eq(gatewayProxy.getConcurrencyLevel().intValue()))) + .thenReturn(mockGateway); + when(mockGatewayHub.getManualStart()).thenReturn(false); + when(mockGateway.getQueueAttributes()).thenReturn(mockGatewayQueueAttributes); + + factoryBean.setBindAddress("10.124.210.42"); + factoryBean.setGateways(Arrays.asList(gatewayProxy)); + factoryBean.setManualStart(false); + //factoryBean.setMaxConnections(50); + factoryBean.setMaximumTimeBetweenPings(20480); + factoryBean.setName(gatewayHubName); + factoryBean.setPort(8484); + factoryBean.setSocketBufferSize(4096); + factoryBean.setStartupPolicy(" PriMary "); + factoryBean.afterPropertiesSet(); + + verify(mockGatewayHub, times(1)).setBindAddress(eq("10.124.210.42")); + verify(mockGatewayHub, times(1)).setManualStart(eq(false)); + //verify(mockGatewayHub, times(1)).setMaxConnections(eq(50)); + verify(mockGatewayHub, times(1)).setMaximumTimeBetweenPings(eq(20480)); + verify(mockGatewayHub, times(1)).setSocketBufferSize(eq(4096)); + verify(mockGatewayHub, times(1)).setStartupPolicy(eq("primary")); + verify(mockGatewayHub, times(1)).addGateway(eq(gatewayProxy.getId()), eq(gatewayProxy.getConcurrencyLevel())); + verify(mockGatewayHub, times(1)).start(); + verify(mockGateway, times(1)).addEndpoint(eq(gatewayEndpointOne.getId()), eq(gatewayEndpointOne.getHost()), + eq(gatewayEndpointOne.getPort())); + verify(mockGateway, times(1)).addEndpoint(eq(gatewayEndpointTwo.getId()), eq(gatewayEndpointTwo.getHost()), + eq(gatewayEndpointTwo.getPort())); + verify(mockGateway, times(1)).addListener(same(mockGatewayListener)); + verify(mockGateway, times(1)).setOrderPolicy(eq(Gateway.OrderPolicy.THREAD)); + verify(mockGateway, times(1)).setSocketBufferSize(eq(gatewayProxy.getSocketBufferSize())); + //verify(mockGateway, times(1)).setSocketReadTimeout(eq(gatewayProxy.getSocketReadTimeout())); + verify(mockGateway, times(1)).getQueueAttributes(); + verify(mockGatewayQueueAttributes, times(1)).setAlertThreshold(eq(gatewayQueue.getAlertThreshold())); + verify(mockGatewayQueueAttributes, times(1)).setBatchConflation(eq(gatewayQueue.getEnableBatchConflation())); + verify(mockGatewayQueueAttributes, times(1)).setBatchSize(eq(gatewayQueue.getBatchSize())); + verify(mockGatewayQueueAttributes, times(1)).setBatchTimeInterval(eq(gatewayQueue.getBatchTimeInterval())); + verify(mockGatewayQueueAttributes, times(1)).setDiskStoreName(eq(gatewayQueue.getDiskStoreRef())); + verify(mockGatewayQueueAttributes, times(1)).setMaximumQueueMemory(eq(gatewayQueue.getMaximumQueueMemory())); + verify(mockGatewayQueueAttributes, times(1)).setEnablePersistence(eq(gatewayQueue.getPersistent())); + } + + @Test + public void testGatewayQueueWithOverflowNoPersistence() throws Exception { + String gatewayHubName = "testGatewayQueueWithOverflowNoPersistence"; + + GatewayProxy.GatewayQueue gatewayQueue = new GatewayProxy.GatewayQueue(); + + gatewayQueue.setAlertThreshold(100); + gatewayQueue.setBatchSize(250); + gatewayQueue.setBatchTimeInterval(120000); + gatewayQueue.setDiskStoreRef("diskZ"); + gatewayQueue.setEnableBatchConflation(true); + gatewayQueue.setMaximumQueueMemory(2048); + gatewayQueue.setPersistent(false); + + GatewayProxy gatewayProxy = new GatewayProxy(); + + gatewayProxy.setId("gatewayProxyId"); + gatewayProxy.setConcurrencyLevel(2); + gatewayProxy.setEndpoints(null); + gatewayProxy.setListeners(null); + gatewayProxy.setOrderPolicy(" thREAD "); + gatewayProxy.setQueue(gatewayQueue); + gatewayProxy.setSocketBufferSize(4096); + //gatewayProxy.setSocketReadTimeout(60); + + GatewayHub mockGatewayHub = mock(GatewayHub.class, "testGatewayQueueWithOverflowNoPersistence.MockGatewayHub"); + + Gateway mockGateway = mock(Gateway.class, "testGatewayQueueWithOverflowNoPersistence.MockGateway"); + + GatewayQueueAttributes mockGatewayQueueAttributes = mock(GatewayQueueAttributes.class, + "testGatewayQueueWithOverflowNoPersistence.MockGatewayQueueAttributes"); + + when(mockCache.addGatewayHub(eq(gatewayHubName), eq(10224))).thenReturn(mockGatewayHub); + when(mockCache.getGatewayHub(eq(gatewayHubName))).thenReturn(mockGatewayHub); + when(mockGatewayHub.addGateway(eq(gatewayProxy.getId()), eq(gatewayProxy.getConcurrencyLevel()))) + .thenReturn(mockGateway); + when(mockGatewayHub.getManualStart()).thenReturn(GatewayHub.DEFAULT_MANUAL_START); + when(mockGateway.getQueueAttributes()).thenReturn(mockGatewayQueueAttributes); + + factoryBean.setGateways(Arrays.asList(gatewayProxy)); + factoryBean.setName(gatewayHubName); + factoryBean.setPort(10224); + factoryBean.afterPropertiesSet(); + + verify(mockGatewayHub, times(1)).setBindAddress(eq(GatewayHub.DEFAULT_BIND_ADDRESS)); + verify(mockGatewayHub, times(1)).setManualStart(eq(GatewayHub.DEFAULT_MANUAL_START)); + verify(mockGatewayHub, times(1)).setMaximumTimeBetweenPings(eq(GatewayHub.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS)); + verify(mockGatewayHub, times(1)).setSocketBufferSize(eq(GatewayHub.DEFAULT_SOCKET_BUFFER_SIZE)); + verify(mockGatewayHub, times(1)).setStartupPolicy(eq(GatewayHub.DEFAULT_STARTUP_POLICY)); + verify(mockGatewayHub, times(1)).start(); + verify(mockGatewayHub, times(1)).addGateway(eq(gatewayProxy.getId()), eq(gatewayProxy.getConcurrencyLevel())); + verify(mockGateway, times(1)).setOrderPolicy(eq(Gateway.OrderPolicy.THREAD)); + verify(mockGateway, times(1)).setSocketBufferSize(eq(gatewayProxy.getSocketBufferSize())); + //verify(mockGateway, times(1)).setSocketReadTimeout(eq(gatewayProxy.getSocketReadTimeout())); + verify(mockGatewayQueueAttributes, times(1)).setAlertThreshold(gatewayQueue.getAlertThreshold()); + verify(mockGatewayQueueAttributes, times(1)).setBatchConflation(gatewayQueue.getEnableBatchConflation()); + verify(mockGatewayQueueAttributes, times(1)).setBatchSize(gatewayQueue.getBatchSize()); + verify(mockGatewayQueueAttributes, times(1)).setBatchTimeInterval(gatewayQueue.getBatchTimeInterval()); + verify(mockGatewayQueueAttributes, times(1)).setDiskStoreName(gatewayQueue.getDiskStoreRef()); + verify(mockGatewayQueueAttributes, times(1)).setMaximumQueueMemory(gatewayQueue.getMaximumQueueMemory()); + verify(mockGatewayQueueAttributes, times(1)).setEnablePersistence(gatewayQueue.getPersistent()); + } + +}