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 7bb42261..61c60ba8 100644
--- a/src/main/java/org/springframework/data/gemfire/wan/AbstractWANComponentFactoryBean.java
+++ b/src/main/java/org/springframework/data/gemfire/wan/AbstractWANComponentFactoryBean.java
@@ -26,32 +26,33 @@ import org.springframework.util.Assert;
import com.gemstone.gemfire.cache.Cache;
/**
- * Base class for Gemfire WAN Gateway component factory beans
+ * Base class for GemFire WAN Gateway component factory beans.
+ *
* @author David Turanski
- *
+ * @author John Blum
*/
public abstract class AbstractWANComponentFactoryBean implements FactoryBean, InitializingBean, BeanNameAware,
DisposableBean {
- protected Log log = LogFactory.getLog(this.getClass());
- private String name;
+ protected Log log = LogFactory.getLog(getClass());
protected final Cache cache;
protected Object factory;
private String beanName;
+ private String name;
- protected AbstractWANComponentFactoryBean(Cache cache) {
+ protected AbstractWANComponentFactoryBean(final Cache cache) {
this.cache = cache;
}
public void setName(String name) {
this.name = name;
}
-
+
public String getName() {
- return name!=null ? name: beanName;
+ return (name != null ? name: beanName);
}
@Override
@@ -88,4 +89,4 @@ public abstract class AbstractWANComponentFactoryBean implements FactoryBean<
this.factory = factory;
}
-}
\ No newline at end of file
+}
diff --git a/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java b/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java
index 0ae7a81a..1750853e 100644
--- a/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java
+++ b/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java
@@ -18,7 +18,6 @@ package org.springframework.data.gemfire.wan;
import java.util.Arrays;
import java.util.List;
-import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.SmartLifecycle;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
@@ -31,55 +30,49 @@ import com.gemstone.gemfire.cache.wan.GatewaySenderFactory;
import com.gemstone.gemfire.cache.wan.GatewayTransportFilter;
/**
- * FactoryBean for creating a GemFire {@link GatewaySender}.
+ * FactoryBean for creating a parallel or serial GemFire {@link GatewaySender}.
+ *
* @author David Turanski
- *
+ * @author John Blum
*/
public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean
-implements SmartLifecycle {
- private static List validOrderPolicyValues = Arrays.asList("KEY", "PARTITION", "THREAD");
+ implements SmartLifecycle {
- private GatewaySender gatewaySender;
+ private static final List VALID_ORDER_POLICIES = Arrays.asList("KEY", "PARTITION", "THREAD");
+
+ private boolean manualStart = false;
private int remoteDistributedSystemId;
+ private GatewaySender gatewaySender;
+
private List eventFilters;
private List transportFilters;
- private Integer alertThreshold;
-
- private Boolean enableBatchConflation;
-
- private Integer batchSize;
-
- private Integer batchTimeInterval;
-
- private String diskStoreRef;
-
private Boolean diskSynchronous;
-
- private Integer dispatcherThreads;
-
- private boolean manualStart = false;
-
- private Integer maximumQueueMemory;
-
- private String orderPolicy;
-
+ private Boolean enableBatchConflation;
private Boolean parallel;
-
private Boolean persistent;
+ private Integer alertThreshold;
+ private Integer batchSize;
+ private Integer batchTimeInterval;
+ private Integer dispatcherThreads;
+ private Integer maximumQueueMemory;
private Integer socketBufferSize;
-
private Integer socketReadTimeout;
+ private String diskStoreRef;
+ private String orderPolicy;
+
/**
- *
- * @param cache the Gemfire cache
+ * Constructs an instance of the GatewaySenderFactoryBean class initialized with a reference to the GemFire cache.
+ *
+ * @param cache the Gemfire cache reference.
+ * @see com.gemstone.gemfire.cache.Cache
*/
- public GatewaySenderFactoryBean(Cache cache) {
+ public GatewaySenderFactoryBean(final Cache cache) {
super(cache);
}
@@ -95,21 +88,18 @@ implements SmartLifecycle {
@Override
protected void doInit() {
- GatewaySenderFactory gatewaySenderFactory = null;
- if (this.factory == null) {
- gatewaySenderFactory = cache.createGatewaySenderFactory();
- } else {
- gatewaySenderFactory = (GatewaySenderFactory)factory;
- }
+ GatewaySenderFactory gatewaySenderFactory = (this.factory != null ? (GatewaySenderFactory) factory :
+ cache.createGatewaySenderFactory());
+
if (diskStoreRef != null) {
- persistent = (persistent == null) ? Boolean.TRUE : persistent;
- Assert.isTrue(persistent, "specifying a disk store requires persistent property to be true");
+ persistent = (persistent == null || persistent);
+ Assert.isTrue(persistent, "Specifying a disk store requires the persistent property to be true.");
gatewaySenderFactory.setDiskStoreName(diskStoreRef);
}
if (diskSynchronous != null) {
- persistent = (persistent == null) ? Boolean.TRUE : persistent;
- Assert.isTrue(persistent, "specifying a disk synchronous requires persistent property to be true");
+ persistent = (persistent == null || persistent);
+ Assert.isTrue(persistent, "Specifying disk synchronous requires the persistent property to be true.");
gatewaySenderFactory.setDiskSynchronous(diskSynchronous);
}
@@ -117,15 +107,15 @@ implements SmartLifecycle {
gatewaySenderFactory.setPersistenceEnabled(persistent);
}
- parallel = (parallel == null) ? Boolean.FALSE : parallel;
-
+ parallel = Boolean.TRUE.equals(parallel);
gatewaySenderFactory.setParallel(parallel);
if (orderPolicy != null) {
- Assert.isTrue(parallel, "specifying an order policy requires the parallel property to be true");
+ Assert.isTrue(isSerialGatewaySender(), "Order Policy cannot be used with a Parallel Gateway Sender Queue.");
+
+ Assert.isTrue(VALID_ORDER_POLICIES.contains(orderPolicy.toUpperCase()),
+ String.format("The value for Order Policy '%1$s' is invalid.", orderPolicy));
- Assert.isTrue(validOrderPolicyValues.contains(orderPolicy.toUpperCase()), "The value of order policy:'"
- + orderPolicy + "' is invalid");
gatewaySenderFactory.setOrderPolicy(Gateway.OrderPolicy.valueOf(orderPolicy.toUpperCase()));
}
@@ -134,11 +124,13 @@ implements SmartLifecycle {
gatewaySenderFactory.addGatewayEventFilter(eventFilter);
}
}
+
if (!CollectionUtils.isEmpty(transportFilters)) {
for (GatewayTransportFilter transportFilter : transportFilters) {
gatewaySenderFactory.addGatewayTransportFilter(transportFilter);
}
}
+
if (alertThreshold != null) {
gatewaySenderFactory.setAlertThreshold(alertThreshold);
}
@@ -152,6 +144,8 @@ implements SmartLifecycle {
gatewaySenderFactory.setBatchTimeInterval(batchTimeInterval);
}
if (dispatcherThreads != null) {
+ Assert.isTrue(isSerialGatewaySender(),
+ "The number of Dispatcher Threads cannot be specified with a Parallel Gateway Sender Queue.");
gatewaySenderFactory.setDispatcherThreads(dispatcherThreads);
}
@@ -166,7 +160,9 @@ implements SmartLifecycle {
if (socketReadTimeout != null) {
gatewaySenderFactory.setSocketReadTimeout(socketReadTimeout);
}
- GatewaySenderWrapper wrapper = new GatewaySenderWrapper(gatewaySenderFactory.create(getName(), remoteDistributedSystemId));
+
+ GatewaySenderWrapper wrapper = new GatewaySenderWrapper(gatewaySenderFactory.create(getName(),
+ remoteDistributedSystemId));
wrapper.setManualStart(manualStart);
gatewaySender = wrapper;
}
@@ -227,6 +223,14 @@ implements SmartLifecycle {
this.parallel = parallel;
}
+ public boolean isSerialGatewaySender() {
+ return !isParallelGatewaySender();
+ }
+
+ public boolean isParallelGatewaySender() {
+ return Boolean.TRUE.equals(parallel);
+ }
+
public void setPersistent(Boolean persistent) {
this.persistent = persistent;
}
@@ -289,4 +293,5 @@ implements SmartLifecycle {
stop();
callback.run();
}
+
}
diff --git a/src/main/resources/META-INF/spring.schemas b/src/main/resources/META-INF/spring.schemas
index d04e7bde..1361f7a9 100644
--- a/src/main/resources/META-INF/spring.schemas
+++ b/src/main/resources/META-INF/spring.schemas
@@ -2,7 +2,9 @@ http\://www.springframework.org/schema/gemfire/spring-gemfire-1.0.xsd=org/spring
http\://www.springframework.org/schema/gemfire/spring-gemfire-1.1.xsd=org/springframework/data/gemfire/config/spring-gemfire-1.1.xsd
http\://www.springframework.org/schema/gemfire/spring-gemfire-1.2.xsd=org/springframework/data/gemfire/config/spring-gemfire-1.2.xsd
http\://www.springframework.org/schema/gemfire/spring-gemfire-1.3.xsd=org/springframework/data/gemfire/config/spring-gemfire-1.3.xsd
-http\://www.springframework.org/schema/gemfire/spring-gemfire.xsd=org/springframework/data/gemfire/config/spring-gemfire-1.3.xsd
+http\://www.springframework.org/schema/gemfire/spring-gemfire-1.3.xsd=org/springframework/data/gemfire/config/spring-gemfire-1.4.xsd
+http\://www.springframework.org/schema/gemfire/spring-gemfire.xsd=org/springframework/data/gemfire/config/spring-gemfire-1.4.xsd
http\://www.springframework.org/schema/data/gemfire/spring-data-gemfire-1.2.xsd=org/springframework/data/gemfire/config/spring-data-gemfire-1.2.xsd
http\://www.springframework.org/schema/data/gemfire/spring-data-gemfire-1.3.xsd=org/springframework/data/gemfire/config/spring-data-gemfire-1.3.xsd
-http\://www.springframework.org/schema/data/gemfire/spring-data-gemfire.xsd=org/springframework/data/gemfire/config/spring-data-gemfire-1.3.xsd
\ No newline at end of file
+http\://www.springframework.org/schema/data/gemfire/spring-data-gemfire-1.4.xsd=org/springframework/data/gemfire/config/spring-data-gemfire-1.4.xsd
+http\://www.springframework.org/schema/data/gemfire/spring-data-gemfire.xsd=org/springframework/data/gemfire/config/spring-data-gemfire-1.4.xsd
diff --git a/src/main/resources/org/springframework/data/gemfire/config/spring-data-gemfire-1.4.xsd b/src/main/resources/org/springframework/data/gemfire/config/spring-data-gemfire-1.4.xsd
new file mode 100644
index 00000000..6a01fad3
--- /dev/null
+++ b/src/main/resources/org/springframework/data/gemfire/config/spring-data-gemfire-1.4.xsd
@@ -0,0 +1,174 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ The reference to a MappingContext. If not set a default one will be created.
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.4.xsd b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.4.xsd
new file mode 100644
index 00000000..610baa87
--- /dev/null
+++ b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.4.xsd
@@ -0,0 +1,3048 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ namespace and its 'properties' element.
+ ]]>
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/src/test/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBeanTest.java b/src/test/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBeanTest.java
new file mode 100644
index 00000000..e103058f
--- /dev/null
+++ b/src/test/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBeanTest.java
@@ -0,0 +1,196 @@
+/*
+ * 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.assertNotNull;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.junit.Test;
+import org.springframework.data.gemfire.TestUtils;
+
+import com.gemstone.gemfire.cache.Cache;
+import com.gemstone.gemfire.cache.util.Gateway;
+import com.gemstone.gemfire.cache.wan.GatewaySender;
+import com.gemstone.gemfire.cache.wan.GatewaySenderFactory;
+
+/**
+ * The GatewaySenderFactoryBeanTest class is a test suite of test cases testing the contract and functionality of the
+ * GatewaySenderFactoryBean class.
+ *
+ * @author John Blum
+ * @see com.gemstone.gemfire.cache.Cache
+ * @see com.gemstone.gemfire.cache.wan.GatewaySender
+ * @see com.gemstone.gemfire.cache.util.Gateway
+ * @see com.gemstone.gemfire.cache.wan.GatewaySenderFactory
+ * @see org.junit.Test
+ * @see org.mockito.Mockito
+ * @see org.springframework.data.gemfire.wan.GatewaySenderFactoryBean
+ * @since 1.4.0
+ */
+public class GatewaySenderFactoryBeanTest {
+
+ protected Cache createMockCacheWithGatewayInfrastructure(final GatewaySenderFactory gatewaySenderFactory) {
+ Cache mockCache = mock(Cache.class);
+ when(mockCache.createGatewaySenderFactory()).thenReturn(gatewaySenderFactory);
+ return mockCache;
+ }
+
+ protected GatewaySenderFactory createMockGatewaySenderFactory(final String gatewaySenderName,
+ final int remoteDistributedSystemId) {
+ GatewaySenderFactory mockGatewaySenderFactory = mock(GatewaySenderFactory.class);
+ GatewaySender mockGatewaySender = mock(GatewaySender.class);
+
+ when(mockGatewaySender.getId()).thenReturn(gatewaySenderName);
+ when(mockGatewaySender.getRemoteDSId()).thenReturn(remoteDistributedSystemId);
+ when(mockGatewaySenderFactory.create(eq(gatewaySenderName), eq(remoteDistributedSystemId)))
+ .thenReturn(mockGatewaySender);
+
+ return mockGatewaySenderFactory;
+ }
+
+ protected void verifyExpectations(final GatewaySenderFactoryBean factoryBean,
+ final GatewaySenderFactory mockGatewaySenderFactory) throws Exception {
+ Boolean parallel = TestUtils.readField("parallel", factoryBean);
+
+ verify(mockGatewaySenderFactory).setParallel(eq(Boolean.TRUE.equals(parallel)));
+
+ String orderPolicy = TestUtils.readField("orderPolicy", factoryBean);
+
+ if (orderPolicy != null) {
+ verify(mockGatewaySenderFactory).setOrderPolicy(eq(Gateway.OrderPolicy.valueOf(orderPolicy.toUpperCase())));
+ }
+
+ Integer dispatcherThreads = TestUtils.readField("dispatcherThreads", factoryBean);
+
+ if (dispatcherThreads != null) {
+ verify(mockGatewaySenderFactory).setDispatcherThreads(eq(dispatcherThreads));
+ }
+ }
+
+ @Test
+ public void testParallelGatewaySender() throws Exception {
+ GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g1", 69);
+
+ GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
+ createMockCacheWithGatewayInfrastructure(mockGatewaySenderFactory));
+
+ factoryBean.setName("g1");
+ factoryBean.setRemoteDistributedSystemId(69);
+ factoryBean.setParallel(true);
+ factoryBean.doInit();
+
+ verifyExpectations(factoryBean, mockGatewaySenderFactory);
+
+ GatewaySender gatewaySender = factoryBean.getObject();
+
+ assertNotNull(gatewaySender);
+ assertEquals("g1", gatewaySender.getId());
+ assertEquals(69, gatewaySender.getRemoteDSId());
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testParallelGatewaySenderWithOrderPolicy() {
+ GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g2", 69);
+
+ GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
+ createMockCacheWithGatewayInfrastructure(mockGatewaySenderFactory));
+
+ factoryBean.setName("g2");
+ factoryBean.setRemoteDistributedSystemId(69);
+ factoryBean.setParallel(true);
+ factoryBean.setOrderPolicy("KEY");
+
+ try {
+ factoryBean.doInit();
+ }
+ catch (IllegalArgumentException expected) {
+ assertEquals("Order Policy cannot be used with a Parallel Gateway Sender Queue.", expected.getMessage());
+ throw expected;
+ }
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testParallelGatewaySenderWithDispatcherThreads() {
+ GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g3", 69);
+
+ GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
+ createMockCacheWithGatewayInfrastructure(mockGatewaySenderFactory));
+
+ factoryBean.setName("g3");
+ factoryBean.setRemoteDistributedSystemId(69);
+ factoryBean.setParallel(true);
+ factoryBean.setDispatcherThreads(1);
+
+ try {
+ factoryBean.doInit();
+ }
+ catch (IllegalArgumentException expected) {
+ assertEquals("The number of Dispatcher Threads cannot be specified with a Parallel Gateway Sender Queue.",
+ expected.getMessage());
+ throw expected;
+ }
+ }
+
+ @Test
+ public void testSerialGatewaySenderWithOrderPolicy() throws Exception {
+ GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g4", 21);
+
+ GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
+ createMockCacheWithGatewayInfrastructure(mockGatewaySenderFactory));
+
+ factoryBean.setName("g4");
+ factoryBean.setRemoteDistributedSystemId(21);
+ factoryBean.setParallel(false);
+ factoryBean.setDispatcherThreads(1);
+ factoryBean.doInit();
+
+ verifyExpectations(factoryBean, mockGatewaySenderFactory);
+
+ GatewaySender gatewaySender = factoryBean.getObject();
+
+ assertNotNull(gatewaySender);
+ assertEquals("g4", gatewaySender.getId());
+ assertEquals(21, gatewaySender.getRemoteDSId());
+ }
+
+ @Test
+ public void testGatewaySenderWithOrderPolicyAndDispatcherThreads() throws Exception {
+ GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g5", 42);
+
+ GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
+ createMockCacheWithGatewayInfrastructure(mockGatewaySenderFactory));
+
+ factoryBean.setName("g5");
+ factoryBean.setRemoteDistributedSystemId(42);
+ factoryBean.setOrderPolicy("THREAD");
+ factoryBean.setDispatcherThreads(1);
+ factoryBean.doInit();
+
+ verifyExpectations(factoryBean, mockGatewaySenderFactory);
+
+ GatewaySender gatewaySender = factoryBean.getObject();
+
+ assertNotNull(gatewaySender);
+ assertEquals("g5", gatewaySender.getId());
+ assertEquals(42, gatewaySender.getRemoteDSId());
+ }
+
+}