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

SGF-865 - Refactor Gateway Sender/Receiver support and remove unnecessary configuration validations.

Browse Source
This commit is contained in:
John Blum
2019-07-23 10:52:55 -07:00
parent ac76726875
commit 094ea58a93
5 changed files with 96 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -13,16 +13,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.gemfire.wan;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.geode.cache.Cache;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.gemfire.support.AbstractFactoryBeanSupport;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -32,13 +31,13 @@ import org.springframework.util.StringUtils;
* @author David Turanski
* @author John Blum
* @see org.apache.geode.cache.Cache
* @see org.springframework.beans.factory.BeanNameAware
* @see org.springframework.beans.factory.DisposableBean
* @see org.springframework.beans.factory.FactoryBean
* @see org.springframework.beans.factory.InitializingBean
* @see org.springframework.data.gemfire.support.AbstractFactoryBeanSupport
*/
public abstract class AbstractWANComponentFactoryBean<T>
implements BeanNameAware, DisposableBean, FactoryBean<T>, InitializingBean {
public abstract class AbstractWANComponentFactoryBean<T> extends AbstractFactoryBeanSupport<T>
implements DisposableBean, InitializingBean {
protected final Cache cache;
@@ -67,12 +66,10 @@ public abstract class AbstractWANComponentFactoryBean<T>
}
public String getName() {
return StringUtils.hasText(this.name) ? this.name : this.beanName;
}
@Override
public final boolean isSingleton() {
return true;
return StringUtils.hasText(this.name)
? this.name
: this.beanName;
}
@Override

61
src/main/java/org/springframework/data/gemfire/wan/GatewayReceiverFactoryBean.java
View File

@@ -15,12 +15,16 @@
*/
package org.springframework.data.gemfire.wan;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.wan.GatewayReceiver;
import org.apache.geode.cache.wan.GatewayReceiverFactory;
import org.apache.geode.cache.wan.GatewayTransportFilter;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.data.gemfire.util.CollectionUtils;
import org.springframework.util.Assert;
@@ -66,42 +70,24 @@ public class GatewayReceiverFactoryBean extends AbstractWANComponentFactoryBean<
super(cache);
}
/**
* @inheritDoc
*/
@Override
public GatewayReceiver getObject() throws Exception {
return gatewayReceiver;
}
protected void doInit() {
/**
* @inheritDoc
*/
@Override
public Class<?> getObjectType() {
return (gatewayReceiver != null ? gatewayReceiver.getClass() : GatewayReceiver.class);
}
GatewayReceiverFactory gatewayReceiverFactory = this.cache.createGatewayReceiverFactory();
/**
* @inheritDoc
*/
@Override
protected void doInit() throws Exception {
GatewayReceiverFactory gatewayReceiverFactory = cache.createGatewayReceiverFactory();
Optional.ofNullable(this.bindAddress)
.filter(StringUtils::hasText)
.ifPresent(gatewayReceiverFactory::setBindAddress);
if (StringUtils.hasText(bindAddress)) {
gatewayReceiverFactory.setBindAddress(bindAddress);
}
if (StringUtils.hasText(hostnameForSenders)) {
gatewayReceiverFactory.setHostnameForSenders(hostnameForSenders);
}
Optional.ofNullable(this.hostnameForSenders)
.filter(StringUtils::hasText)
.ifPresent(gatewayReceiverFactory::setHostnameForSenders);
int localStartPort = defaultPort(startPort, GatewayReceiver.DEFAULT_START_PORT);
int localEndPort = defaultPort(endPort, GatewayReceiver.DEFAULT_END_PORT);
Assert.isTrue(localStartPort <= localEndPort,
String.format("'startPort' must be less than or equal to %d.", localEndPort));
String.format("[startPort] must be less than or equal to [%d]", localEndPort));
gatewayReceiverFactory.setStartPort(localStartPort);
gatewayReceiverFactory.setEndPort(localEndPort);
@@ -115,13 +101,24 @@ public class GatewayReceiverFactoryBean extends AbstractWANComponentFactoryBean<
gatewayReceiverFactory.setSocketBufferSize(socketBufferSize);
}
for (GatewayTransportFilter transportFilter : CollectionUtils.nullSafeList(transportFilters)) {
gatewayReceiverFactory.addGatewayTransportFilter(transportFilter);
}
CollectionUtils.nullSafeList(this.transportFilters).forEach(gatewayReceiverFactory::addGatewayTransportFilter);
gatewayReceiver = gatewayReceiverFactory.create();
}
@Override
public GatewayReceiver getObject() throws Exception {
return this.gatewayReceiver;
}
@Override
public Class<?> getObjectType() {
return this.gatewayReceiver != null
? this.gatewayReceiver.getClass()
: GatewayReceiver.class;
}
protected int defaultPort(Integer port, int defaultPort) {
return (port != null ? port : defaultPort);
}
@@ -161,4 +158,8 @@ public class GatewayReceiverFactoryBean extends AbstractWANComponentFactoryBean<
public void setTransportFilters(List<GatewayTransportFilter> transportFilters) {
this.transportFilters = transportFilters;
}
public Collection<? extends GatewayTransportFilter> getTransportFilters() {
return Collections.unmodifiableList(this.transportFilters);
}
}

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

@@ -16,6 +16,7 @@
package org.springframework.data.gemfire.wan;
import java.util.List;
import java.util.Optional;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.wan.GatewayEventFilter;
@@ -23,9 +24,10 @@ import org.apache.geode.cache.wan.GatewayEventSubstitutionFilter;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.cache.wan.GatewaySenderFactory;
import org.apache.geode.cache.wan.GatewayTransportFilter;
import org.apache.shiro.util.StringUtils;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.data.gemfire.util.CollectionUtils;
import org.springframework.util.Assert;
/**
* Spring {@link FactoryBean} for creating a parallel or serial Pivotal GemFire {@link GatewaySender}.
@@ -88,79 +90,50 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
@Override
protected void doInit() {
GatewaySenderFactory gatewaySenderFactory = this.factory != null
? (GatewaySenderFactory) this.factory
: this.cache.createGatewaySenderFactory();
GatewaySenderFactory gatewaySenderFactory = resolveGatewaySenderFactory();
if (alertThreshold != null) {
gatewaySenderFactory.setAlertThreshold(alertThreshold);
}
Optional.ofNullable(this.alertThreshold).ifPresent(gatewaySenderFactory::setAlertThreshold);
Optional.ofNullable(this.batchConflationEnabled).ifPresent(gatewaySenderFactory::setBatchConflationEnabled);
Optional.ofNullable(this.batchSize).ifPresent(gatewaySenderFactory::setBatchSize);
Optional.ofNullable(this.batchTimeInterval).ifPresent(gatewaySenderFactory::setBatchTimeInterval);
if (batchConflationEnabled != null) {
gatewaySenderFactory.setBatchConflationEnabled(batchConflationEnabled);
}
Optional.ofNullable(this.diskStoreReference)
.filter(StringUtils::hasText)
.ifPresent(gatewaySenderFactory::setDiskStoreName);
if (batchSize != null) {
gatewaySenderFactory.setBatchSize(batchSize);
}
Optional.ofNullable(this.diskSynchronous).ifPresent(gatewaySenderFactory::setDiskSynchronous);
Optional.ofNullable(this.dispatcherThreads).ifPresent(gatewaySenderFactory::setDispatcherThreads);
if (batchTimeInterval != null) {
gatewaySenderFactory.setBatchTimeInterval(batchTimeInterval);
}
CollectionUtils.nullSafeList(this.eventFilters).forEach(gatewaySenderFactory::addGatewayEventFilter);
if (diskStoreReference != null) {
gatewaySenderFactory.setDiskStoreName(diskStoreReference);
}
Optional.ofNullable(this.eventSubstitutionFilter)
.ifPresent(gatewaySenderFactory::setGatewayEventSubstitutionFilter);
if (diskSynchronous != null) {
gatewaySenderFactory.setDiskSynchronous(diskSynchronous);
}
gatewaySenderFactory.setManualStart(this.manualStart);
if (dispatcherThreads != null) {
gatewaySenderFactory.setDispatcherThreads(dispatcherThreads);
}
for (GatewayEventFilter eventFilter : CollectionUtils.nullSafeList(eventFilters)) {
gatewaySenderFactory.addGatewayEventFilter(eventFilter);
}
if (eventSubstitutionFilter != null) {
gatewaySenderFactory.setGatewayEventSubstitutionFilter(eventSubstitutionFilter);
}
gatewaySenderFactory.setManualStart(manualStart);
if (maximumQueueMemory != null) {
gatewaySenderFactory.setMaximumQueueMemory(maximumQueueMemory);
}
if (orderPolicy != null) {
Assert.isTrue(isSerialGatewaySender(), "OrderPolicy cannot be used with a Parallel GatewaySender");
gatewaySenderFactory.setOrderPolicy(this.orderPolicy);
}
Optional.ofNullable(this.maximumQueueMemory).ifPresent(gatewaySenderFactory::setMaximumQueueMemory);
Optional.ofNullable(this.orderPolicy).ifPresent(gatewaySenderFactory::setOrderPolicy);
gatewaySenderFactory.setParallel(isParallelGatewaySender());
gatewaySenderFactory.setPersistenceEnabled(isPersistent());
if (socketBufferSize != null) {
gatewaySenderFactory.setSocketBufferSize(socketBufferSize);
}
Optional.ofNullable(this.socketBufferSize).ifPresent(gatewaySenderFactory::setSocketBufferSize);
Optional.ofNullable(this.socketReadTimeout).ifPresent(gatewaySenderFactory::setSocketReadTimeout);
if (socketReadTimeout != null) {
gatewaySenderFactory.setSocketReadTimeout(socketReadTimeout);
}
for (GatewayTransportFilter transportFilter : CollectionUtils.nullSafeList(transportFilters)) {
gatewaySenderFactory.addGatewayTransportFilter(transportFilter);
}
CollectionUtils.nullSafeList(this.transportFilters).forEach(gatewaySenderFactory::addGatewayTransportFilter);
GatewaySenderWrapper wrapper =
new GatewaySenderWrapper(gatewaySenderFactory.create(getName(), remoteDistributedSystemId));
new GatewaySenderWrapper(gatewaySenderFactory.create(getName(), this.remoteDistributedSystemId));
wrapper.setManualStart(manualStart);
gatewaySender = wrapper;
wrapper.setManualStart(this.manualStart);
this.gatewaySender = wrapper;
}
private GatewaySenderFactory resolveGatewaySenderFactory() {
return this.factory != null
? (GatewaySenderFactory) this.factory
: this.cache.createGatewaySenderFactory();
}
@Override
@@ -170,7 +143,10 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
@Override
public Class<?> getObjectType() {
return this.gatewaySender != null ? this.gatewaySender.getClass() : GatewaySender.class;
return this.gatewaySender != null
? this.gatewaySender.getClass()
: GatewaySender.class;
}
public void setAlertThreshold(Integer alertThreshold) {
@@ -217,7 +193,7 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
this.eventFilters = gatewayEventFilters;
}
public void setEventSubstitutionFilter(final GatewayEventSubstitutionFilter eventSubstitutionFilter) {
public void setEventSubstitutionFilter(GatewayEventSubstitutionFilter eventSubstitutionFilter) {
this.eventSubstitutionFilter = eventSubstitutionFilter;
}
@@ -242,7 +218,7 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
}
public boolean isParallelGatewaySender() {
return Boolean.TRUE.equals(parallel);
return Boolean.TRUE.equals(this.parallel);
}
public void setPersistent(Boolean persistent) {

27
src/test/java/org/springframework/data/gemfire/wan/GatewayReceiverFactoryBeanTest.java
View File

@@ -13,28 +13,28 @@
* 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.assertj.core.api.Assertions.assertThat;
import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.same;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.Arrays;
import java.util.Collections;
import org.junit.Test;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.wan.GatewayReceiverFactory;
import org.apache.geode.cache.wan.GatewayTransportFilter;
import org.junit.Test;
/**
* The GatewayReceiverFactoryBeanTest class is a test suite of test cases testing the contract and functionality
* of the GatewayReceiverFactoryBean class.
* Unit Tests for {@link GatewayReceiverFactoryBean}.
*
* @author John Blum
* @see org.junit.Test
* @see org.springframework.data.gemfire.wan.GatewayReceiverFactoryBean
* @see org.apache.geode.cache.wan.GatewayReceiverFactory
* @since 1.5.0
@@ -42,9 +42,12 @@ import org.junit.Test;
public class GatewayReceiverFactoryBeanTest {
@Test
public void testDoInit() throws Exception {
public void doInitConfiguresGatewaySenderFactoryBean() throws Exception {
Cache mockCache = mock(Cache.class, "testDoInit.Cache");
GatewayReceiverFactory mockGatewayReceiverFactory = mock(GatewayReceiverFactory.class, "testDoInit.GatewayReceiverFactory");
GatewayTransportFilter mockGatewayTransportFilter = mock(GatewayTransportFilter.class, "testDoInit.GatewayTransportFilter");
when(mockCache.createGatewayReceiverFactory()).thenReturn(mockGatewayReceiverFactory);
@@ -59,7 +62,7 @@ public class GatewayReceiverFactoryBeanTest {
factoryBean.setMaximumTimeBetweenPings(5000);
factoryBean.setName("testDoInit");
factoryBean.setSocketBufferSize(16384);
factoryBean.setTransportFilters(Arrays.asList(mockGatewayTransportFilter));
factoryBean.setTransportFilters(Collections.singletonList(mockGatewayTransportFilter));
factoryBean.afterPropertiesSet();
verify(mockGatewayReceiverFactory).setBindAddress(eq("10.224.112.77"));
@@ -74,7 +77,7 @@ public class GatewayReceiverFactoryBeanTest {
}
@Test(expected = IllegalArgumentException.class)
public void testDoInitWithIllegalStartEndPorts() throws Exception {
public void doInitWithIllegalStartEndPorts() throws Exception {
try {
Cache mockCache = mock(Cache.class, "testDoInitWithIllegalStartEndPorts.Cache");
GatewayReceiverFactory mockGatewayReceiverFactory = mock(GatewayReceiverFactory.class,
@@ -90,9 +93,11 @@ public class GatewayReceiverFactoryBeanTest {
factoryBean.afterPropertiesSet();
}
catch (IllegalArgumentException expected) {
assertEquals("'startPort' must be less than or equal to 8192.", expected.getMessage());
assertThat(expected).hasMessageContaining("[startPort] must be less than or equal to [8192]");
assertThat(expected).hasNoCause();
throw expected;
}
}
}

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

@@ -13,7 +13,6 @@
* 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;
@@ -24,10 +23,12 @@ import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import org.junit.Test;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.cache.wan.GatewaySenderFactory;
import org.junit.Test;
import org.springframework.data.gemfire.TestUtils;
/**
@@ -181,29 +182,6 @@ public class GatewaySenderFactoryBeanTest {
assertEquals(69, gatewaySender.getRemoteDSId());
}
@Test(expected = IllegalArgumentException.class)
public void parallelGatewaySenderWithOrderPolicyCreation() {
GatewaySenderFactory mockGatewaySenderFactory =
mockGatewaySenderFactory("g2", 69);
GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
mockCacheWithGatewayInfrastructure(mockGatewaySenderFactory));
factoryBean.setName("g2");
factoryBean.setRemoteDistributedSystemId(69);
factoryBean.setParallel(true);
factoryBean.setOrderPolicy(GatewaySender.OrderPolicy.KEY);
try {
factoryBean.doInit();
}
catch (IllegalArgumentException expected) {
assertEquals("OrderPolicy cannot be used with a Parallel GatewaySender", expected.getMessage());
throw expected;
}
}
@Test
public void gatewaySenderCreationWithDiskSynchronousAndNoPersistence() throws Exception {