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

SGF-398 - Provide early support of Apache Geode (Pivotal GemFire OSS).

Browse Source 
Fixed up and polished the GatewaySenderFactoryBean 'orderPolicy' configuration and validation changing the loosely, String-typed 'orderPolicy' property to use the GatewaySender.OrderPolicy enum.
This commit is contained in:
John Blum
2015-05-01 11:29:51 -07:00
parent aa706b025c
commit abbbfed8fd
3 changed files with 135 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -59,6 +59,8 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
private Boolean parallel;
private Boolean persistent;
private GatewaySender.OrderPolicy orderPolicy;
private GatewayEventSubstitutionFilter eventSubstitutionFilter;
private Integer alertThreshold;
@@ -70,7 +72,6 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
private Integer socketReadTimeout;
private String diskStoreReference;
private String orderPolicy;
/**
* Constructs an instance of the GatewaySenderFactoryBean class initialized with a reference to the GemFire cache.
@@ -101,6 +102,10 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
gatewaySenderFactory.setAlertThreshold(alertThreshold);
}
if (batchConflationEnabled != null) {
gatewaySenderFactory.setBatchConflationEnabled(batchConflationEnabled);
}
if (batchSize != null) {
gatewaySenderFactory.setBatchSize(batchSize);
}
@@ -121,10 +126,6 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
gatewaySenderFactory.setDispatcherThreads(dispatcherThreads);
}
if (batchConflationEnabled != null) {
gatewaySenderFactory.setBatchConflationEnabled(batchConflationEnabled);
}
if (!CollectionUtils.isEmpty(eventFilters)) {
for (GatewayEventFilter eventFilter : eventFilters) {
gatewaySenderFactory.addGatewayEventFilter(eventFilter);
@@ -143,11 +144,7 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
if (orderPolicy != null) {
Assert.isTrue(isSerialGatewaySender(), "Order Policy cannot be used with a Parallel Gateway Sender Queue.");
Assert.isTrue(VALID_ORDER_POLICIES.contains(orderPolicy.toUpperCase()),
String.format("The value for Order Policy '%1$s' is invalid.", orderPolicy));
gatewaySenderFactory.setOrderPolicy(GatewaySender.OrderPolicy.valueOf(orderPolicy.toUpperCase()));
gatewaySenderFactory.setOrderPolicy(orderPolicy);
}
gatewaySenderFactory.setParallel(isParallelGatewaySender());
@@ -230,7 +227,16 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
this.maximumQueueMemory = maximumQueueMemory;
}
/**
* @see #setOrderPolicy(com.gemstone.gemfire.cache.wan.GatewaySender.OrderPolicy)
* @deprecated please use setOrderPolicy(:GatewaySender.OrderPolicy) instead.
*/
@Deprecated
public void setOrderPolicy(String orderPolicy) {
setOrderPolicy(new OrderPolicyConverter().convert(orderPolicy));
}
public void setOrderPolicy(GatewaySender.OrderPolicy orderPolicy) {
this.orderPolicy = orderPolicy;
}
@@ -274,7 +280,8 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
this.transportFilters = gatewayTransportFilters;
}
/* (non-Javadoc)
/*
* (non-Javadoc)
* @see org.springframework.context.SmartLifecycle#isAutoStartup()
*/
@Override
@@ -282,7 +289,8 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
return !manualStart;
}
/* (non-Javadoc)
/*
* (non-Javadoc)
* @see org.springframework.context.Phased#getPhase()
*/
@Override
@@ -290,7 +298,8 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
return Integer.MAX_VALUE;
}
/* (non-Javadoc)
/*
* (non-Javadoc)
* @see org.springframework.context.Lifecycle#isRunning()
*/
@Override
@@ -298,7 +307,8 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
return gatewaySender.isRunning();
}
/* (non-Javadoc)
/*
* (non-Javadoc)
* @see org.springframework.context.Lifecycle#start()
*/
@Override
@@ -310,7 +320,8 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
}
}
/* (non-Javadoc)
/*
* (non-Javadoc)
* @see org.springframework.context.Lifecycle#stop()
*/
@Override
@@ -318,7 +329,8 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
gatewaySender.stop();
}
/* (non-Javadoc)
/*
* (non-Javadoc)
* @see org.springframework.context.SmartLifecycle#stop(java.lang.Runnable)
*/
@Override

22
src/test/java/org/springframework/data/gemfire/TestUtils.java
View File

@@ -25,21 +25,27 @@ public abstract class TestUtils {
@SuppressWarnings("unchecked")
public static <T> T readField(String name, Object target) throws Exception {
Class<?> targetType = target.getClass();
Field field = null;
Class<?> clazz = target.getClass();
do {
try {
field = clazz.getDeclaredField(name);
} catch (Exception ex) {
field = targetType.getDeclaredField(name);
}
catch (Exception ignore) {
}
clazz = clazz.getSuperclass();
} while (field == null && !clazz.equals(Object.class));
targetType = targetType.getSuperclass();
}
while (field == null && !Object.class.equals(targetType));
if (field == null) {
throw new IllegalArgumentException(String.format("Cannot find field '%1$s' in the class hierarchy of %2$s!",
name, targetType));
}
if (field == null)
throw new IllegalArgumentException("Cannot find field '" + name + "' in the class hierarchy of "
+ target.getClass());
field.setAccessible(true);
return (T) field.get(target);
}

159
src/test/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBeanTest.java
View File

@@ -16,8 +16,10 @@
package org.springframework.data.gemfire.wan;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -36,12 +38,12 @@ import com.gemstone.gemfire.cache.wan.GatewaySenderFactory;
* GatewaySenderFactoryBean class.
*
* @author John Blum
* @see com.gemstone.gemfire.cache.Cache
* @see com.gemstone.gemfire.cache.wan.GatewaySender
* @see com.gemstone.gemfire.cache.wan.GatewaySenderFactory
* @see org.junit.Test
* @see org.mockito.Mockito
* @see org.springframework.data.gemfire.wan.GatewaySenderFactoryBean
* @see com.gemstone.gemfire.cache.Cache
* @see com.gemstone.gemfire.cache.wan.GatewaySender
* @see com.gemstone.gemfire.cache.wan.GatewaySenderFactory
* @since 1.4.0
*/
public class GatewaySenderFactoryBeanTest {
@@ -71,11 +73,10 @@ public class GatewaySenderFactoryBeanTest {
verify(mockGatewaySenderFactory).setParallel(eq(Boolean.TRUE.equals(parallel)));
String orderPolicy = TestUtils.readField("orderPolicy", factoryBean);
GatewaySender.OrderPolicy orderPolicy = TestUtils.readField("orderPolicy", factoryBean);
if (orderPolicy != null) {
verify(mockGatewaySenderFactory).setOrderPolicy(eq(GatewaySender.OrderPolicy.valueOf(
orderPolicy.toUpperCase())));
verify(mockGatewaySenderFactory).setOrderPolicy(eq(orderPolicy));
}
Integer dispatcherThreads = TestUtils.readField("dispatcherThreads", factoryBean);
@@ -107,7 +108,7 @@ public class GatewaySenderFactoryBeanTest {
}
@Test
public void testConcurrentParallelGatewaySender() throws Exception {
public void concurrentParallelGatewaySenderCreation() throws Exception {
GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g0", 69);
GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
@@ -129,49 +130,7 @@ public class GatewaySenderFactoryBeanTest {
}
@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
public void testConcurrentSerialGatewaySender() throws Exception {
public void concurrentSerialGatewaySenderCreation() throws Exception {
GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g4", 21);
GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
@@ -193,16 +152,15 @@ public class GatewaySenderFactoryBeanTest {
}
@Test
public void testGatewaySenderWithOrderPolicyAndDispatcherThreads() throws Exception {
GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g5", 42);
public void parallelGatewaySenderCreation() throws Exception {
GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g1", 69);
GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
createMockCacheWithGatewayInfrastructure(mockGatewaySenderFactory));
factoryBean.setName("g5");
factoryBean.setRemoteDistributedSystemId(42);
factoryBean.setOrderPolicy("THREAD");
factoryBean.setDispatcherThreads(1);
factoryBean.setName("g1");
factoryBean.setRemoteDistributedSystemId(69);
factoryBean.setParallel(true);
factoryBean.doInit();
verifyExpectations(factoryBean, mockGatewaySenderFactory);
@@ -210,12 +168,54 @@ public class GatewaySenderFactoryBeanTest {
GatewaySender gatewaySender = factoryBean.getObject();
assertNotNull(gatewaySender);
assertEquals("g5", gatewaySender.getId());
assertEquals(42, gatewaySender.getRemoteDSId());
assertEquals("g1", gatewaySender.getId());
assertEquals(69, gatewaySender.getRemoteDSId());
}
@Test(expected = IllegalArgumentException.class)
public void parallelGatewaySenderWithOrderPolicyCreation() {
GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g2", 69);
GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
createMockCacheWithGatewayInfrastructure(mockGatewaySenderFactory));
factoryBean.setName("g2");
factoryBean.setRemoteDistributedSystemId(69);
factoryBean.setParallel(true);
factoryBean.setOrderPolicy(GatewaySender.OrderPolicy.KEY);
try {
factoryBean.doInit();
}
catch (IllegalArgumentException expected) {
assertEquals("Order Policy cannot be used with a Parallel Gateway Sender Queue.", expected.getMessage());
throw expected;
}
}
@Test
public void testGatewaySenderWithOverflowDiskStoreNoPersistence() throws Exception {
public void gatewaySenderCreationWithDiskSynchronousAndNoPersistence() throws Exception {
GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g7", 51);
GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
createMockCacheWithGatewayInfrastructure(mockGatewaySenderFactory));
factoryBean.setName("g7");
factoryBean.setRemoteDistributedSystemId(51);
factoryBean.setDiskSynchronous(true);
factoryBean.doInit();
verifyExpectations(factoryBean, mockGatewaySenderFactory);
GatewaySender gatewaySender = factoryBean.getObject();
assertNotNull(gatewaySender);
assertEquals("g7", gatewaySender.getId());
assertEquals(51, gatewaySender.getRemoteDSId());
}
@Test
public void gatewaySenderCreationWithOverflowDiskStoreAndNoPersistence() throws Exception {
GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g6", 51);
GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
@@ -237,15 +237,16 @@ public class GatewaySenderFactoryBeanTest {
}
@Test
public void testGatewaySenderWithDiskSynchronousSetPersistenceUnset() throws Exception {
GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g7", 51);
public void gatewaySenderCreationWithOrderPolicyAndDispatcherThreads() throws Exception {
GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g5", 42);
GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
createMockCacheWithGatewayInfrastructure(mockGatewaySenderFactory));
factoryBean.setName("g7");
factoryBean.setRemoteDistributedSystemId(51);
factoryBean.setDiskSynchronous(true);
factoryBean.setName("g5");
factoryBean.setRemoteDistributedSystemId(42);
factoryBean.setOrderPolicy(GatewaySender.OrderPolicy.THREAD);
factoryBean.setDispatcherThreads(1);
factoryBean.doInit();
verifyExpectations(factoryBean, mockGatewaySenderFactory);
@@ -253,8 +254,34 @@ public class GatewaySenderFactoryBeanTest {
GatewaySender gatewaySender = factoryBean.getObject();
assertNotNull(gatewaySender);
assertEquals("g7", gatewaySender.getId());
assertEquals(51, gatewaySender.getRemoteDSId());
assertEquals("g5", gatewaySender.getId());
assertEquals(42, gatewaySender.getRemoteDSId());
}
@Test
@SuppressWarnings("deprecation")
public void setOrderPolicyAsValidString() throws Exception {
GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(mock(Cache.class,
"setOrderPolicyAsAValidString.MockCache"));
factoryBean.setOrderPolicy(GatewaySender.OrderPolicy.THREAD.name().toLowerCase());
assertThat(TestUtils.<GatewaySender.OrderPolicy>readField("orderPolicy", factoryBean),
is(GatewaySender.OrderPolicy.THREAD));
}
@Test(expected = IllegalArgumentException.class)
@SuppressWarnings("deprecation")
public void setOrderPolicyAsInvalidString() {
try {
new GatewaySenderFactoryBean(mock(Cache.class, "setOrderPolicyAsInvalidString.MockCache"))
.setOrderPolicy("invalid");
}
catch (IllegalArgumentException expected) {
assertEquals(String.format("(invalid) is not a valid %1$s!", GatewaySender.OrderPolicy.class.getSimpleName()),
expected.getMessage());
throw expected;
}
}
}