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

SGF-376 - The GemFire WAN GatewayHub support needs refactoring and test coverage.

Browse Source
This commit is contained in:
John Blum
2015-02-25 15:18:42 -08:00
parent b8b6918104
commit 57a47844ab
6 changed files with 715 additions and 343 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
src/main/java/org/springframework/data/gemfire/config/GatewayHubParser.java
View File

@@ -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<Element> gatewayElements = DomUtils.getChildElementsByTagName(element, "gateway");
if (!CollectionUtils.isEmpty(gatewayElements)) {
ManagedList gateways = new ManagedList();
ManagedList<BeanDefinition> gateways = new ManagedList<BeanDefinition>();
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<Element> 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<Element> endpointElements = DomUtils.getChildElementsByTagName(gatewayElement, "gateway-endpoint");
if (!CollectionUtils.isEmpty(endpointElements)) {
ManagedList<BeanDefinition> endpoints = new ManagedList<BeanDefinition>();
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());
}
}
}

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

@@ -35,8 +35,8 @@ import com.gemstone.gemfire.management.internal.cli.util.spring.StringUtils;
* @author David Turanski
* @author John Blum
*/
public abstract class AbstractWANComponentFactoryBean<T> implements BeanNameAware, FactoryBean<T>, InitializingBean,
DisposableBean {
public abstract class AbstractWANComponentFactoryBean<T> implements BeanNameAware, FactoryBean<T>,
InitializingBean, DisposableBean {
protected static final List<String> VALID_ORDER_POLICIES = Arrays.asList("KEY", "PARTITION", "THREAD");
@@ -67,7 +67,7 @@ public abstract class AbstractWANComponentFactoryBean<T> 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<T> 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();
}

266
src/main/java/org/springframework/data/gemfire/wan/GatewayHubFactoryBean.java
View File

@@ -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<GatewayHub> {
private static List<String> validStartupPolicyValues = Arrays.asList("none", "primary", "secondary");
private static List<String> validOrderPolicyValues = Arrays.asList("KEY,PARTITION,THREAD");
private static List<String> 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<String> 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<GatewayProxy> 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<Gatew
@Override
public Class<?> 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<GatewayProxy> 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<GatewayProxy> gateways) {
this.gateways = gateways;
}
List<GatewayProxy> getGateways() {
return (gateways != null ? gateways : Collections.<GatewayProxy>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);
}
}

181
src/main/java/org/springframework/data/gemfire/wan/GatewayProxy.java
View File

@@ -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<GatewayEndpoint> endpoints;
private Integer concurrencyLevel = Gateway.DEFAULT_CONCURRENCY_LEVEL;
private String id;
private List<GatewayEventListener> 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<GatewayEndpoint> endpoints) {
this.endpoints = endpoints;
}
public List<GatewayEndpoint> getEndpoints() {
return (endpoints != null ? endpoints : Collections.<GatewayEndpoint>emptyList());
}
public void setId(String id) {
this.id = id;
}
public String getId() {
return id;
}
public void setListeners(List<GatewayEventListener> listeners) {
this.listeners = listeners;
}
public List<GatewayEventListener> getListeners() {
return (listeners != null ? listeners : Collections.<GatewayEventListener>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<GatewayEndpoint> getEndpoints() {
return endpoints;
}
public String getId() {
return this.id;
}
public List<GatewayEventListener> 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);
}
}
}

136
src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.5.xsd
View File

@@ -2894,7 +2894,7 @@ The id of the cache - default is gemfireCache
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<!-- -->
<xsd:element name="gateway-receiver" type="gatewayReceiverType" />
<!-- -->
<xsd:element name="function-service">
@@ -2957,7 +2957,7 @@ Scope also determines whether to allow remote invocation of some of the region
<xsd:enumeration value="global" />
</xsd:restriction>
</xsd:simpleType>
<!-- -->
<xsd:complexType name="gatewayTransportFilterType">
<xsd:annotation>
<xsd:documentation
@@ -2989,7 +2989,7 @@ use inner bean declarations.
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<!-- Gemfire 6 WAN Gateway schema -->
<!-- GemFire 6 WAN Gateway schema -->
<xsd:complexType name="gatewayHubType">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -2997,8 +2997,7 @@ Deprecated as of Gemfire 7
]]></xsd:documentation>
</xsd:annotation>
<xsd:sequence>
<xsd:element name="gateway" type="gatewayType" minOccurs="0"
maxOccurs="unbounded">
<xsd:element name="gateway" type="gatewayType" minOccurs="0" maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation><![CDATA[
Deprecated as of Gemfire 7
@@ -3013,13 +3012,6 @@ The id of this hub
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="port" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The port for this hub (integer value, if not specified, Gemfire will select an open port)
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="cache-ref" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -3027,32 +3019,35 @@ The id of the cache - default is gemfireCache
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="bind-address" type="xsd:string"
use="optional">
<xsd:attribute name="bind-address" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the bind address (IP address or host name) for the gateway hub
]]></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="manual-start" type="xsd:string"
use="optional">
<xsd:attribute name="manual-start" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies if the gateway hub is manually (true) or automatically(false) started
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="startup-policy" type="xsd:string"
use="optional">
<xsd:attribute name="port" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The port for this hub (integer value, if not specified, Gemfire will select an open port)
]]></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="startup-policy" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the startup policy (primary,secondary, none) for the gateway hub
@@ -3067,12 +3062,9 @@ Specifies the startup policy (primary,secondary, none) for the gateway hub
Deprecated as of Gemfire 7
]]></xsd:documentation>
</xsd:annotation>
<xsd:attribute name="host" type="xsd:string" use="required">
</xsd:attribute>
<xsd:attribute name="port" type="xsd:string" use="required">
</xsd:attribute>
<xsd:attribute name="endpoint-id" type="xsd:string" use="required">
</xsd:attribute>
<xsd:attribute name="endpoint-id" type="xsd:string" use="required"/>
<xsd:attribute name="host" type="xsd:string" use="required"/>
<xsd:attribute name="port" type="xsd:string" use="required"/>
</xsd:complexType>
<!-- -->
<xsd:complexType name="gatewayQueueType">
@@ -3081,24 +3073,7 @@ Deprecated as of Gemfire 7
Deprecated as of Gemfire 7
]]></xsd:documentation>
</xsd:annotation>
<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="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.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="alert-threshold" type="xsd:string"
use="optional">
<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
@@ -3113,26 +3088,38 @@ Specifies the batch size
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="persistent" type="xsd:string" use="optional">
<xsd:attribute name="batch-time-interval" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies whether persistence is enabled: true or false(default)
The maximum time interval that can elapse before a partial batch is sent from a GatewaySender to its corresponding GatewayReceiver.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="disk-store-ref" type="xsd:string"
use="optional">
<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: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="maximum-queue-memory" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the maximum memory in MB to allocate for the queue
]]></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)
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -3146,7 +3133,7 @@ Deprecated as of Gemfire 7
</xsd:annotation>
<xsd:sequence>
<xsd:choice>
<xsd:element name="gateway-endpoint" minOccurs="1" maxOccurs="unbounded" type="gatewayEndpointType"/>
<xsd:element name="gateway-endpoint" type="gatewayEndpointType" minOccurs="1" maxOccurs="unbounded"/>
<xsd:element name="gateway-listener" minOccurs="1" maxOccurs="1">
<xsd:annotation>
<xsd:documentation
@@ -3161,8 +3148,7 @@ An gateway event listener definition for the gateway
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:any namespace="##other" processContents="skip"
minOccurs="0" maxOccurs="unbounded">
<xsd:any namespace="##other" processContents="skip" minOccurs="0" maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation><![CDATA[
Inner bean definition of the gateway event listener
@@ -3181,8 +3167,7 @@ use inner bean declarations.
</xsd:complexType>
</xsd:element>
</xsd:choice>
<xsd:element name="gateway-queue" minOccurs="0"
maxOccurs="1" type="gatewayQueueType" />
<xsd:element name="gateway-queue" type="gatewayQueueType" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="gateway-id" type="xsd:string" use="required">
<xsd:annotation>
@@ -3191,24 +3176,14 @@ Specifies the id for this gateway
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="socket-buffer-size" type="xsd:string"
use="optional">
<xsd:attribute name="concurrency-level" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the socket buffer size in bytes
Specifies the number of parallel threads
]]></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:attribute name="order-policy" type="xsd:string"
use="optional">
<xsd:attribute name="order-policy" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the order policy - This only applies if parallel is enabled:
@@ -3218,11 +3193,17 @@ THREAD:Indicates that events will be parallelized based on the event's originati
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="concurrency-level" type="xsd:string"
use="optional">
<xsd:attribute name="socket-buffer-size" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the number of parallel threads
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>
@@ -3236,7 +3217,6 @@ Deprecated as of Gemfire 7
</xsd:annotation>
</xsd:element>
<!-- End Gemfire 6 WAN Gateway schema -->
<!-- Function Annotation Support -->
<xsd:element name="annotation-driven">
<xsd:annotation>

325
src/test/java/org/springframework/data/gemfire/wan/GatewayHubFactoryBeanTest.java Normal file
View File

@@ -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<GatewayProxy> 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());
}
}