SGF-199 fixed GatewaySender manual-start

This commit is contained in:
David Turanski
2013-09-16 18:35:36 -04:00
parent fb4e184794
commit a4b6acff07
7 changed files with 234 additions and 51 deletions

2
.gitignore vendored
View File

@@ -12,6 +12,8 @@ pom.xml
.classpath
.project
.settings/
.idea/
_site/
/samples/hello-world/vf.gf.dmn-events.cfg
/samples/hello-world/vf.gf.dmn-license.cfg

View File

@@ -24,6 +24,7 @@ import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.io.Resource;
import org.springframework.data.gemfire.client.ClientRegionFactoryBean;
import org.springframework.data.gemfire.wan.GatewaySenderWrapper;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import org.springframework.util.ReflectionUtils;
@@ -47,12 +48,12 @@ import com.gemstone.gemfire.cache.wan.GatewaySender;
* Base class for FactoryBeans used to create GemFire {@link Region}s. Will try
* to first locate the region (by name) and, in case none if found, proceed to
* creating one using the given settings.
*
*
* Note that this factory bean allows for very flexible creation of GemFire
* {@link Region}. For "client" regions however, see
* {@link ClientRegionFactoryBean} which offers easier configuration and
* defaults.
*
*
* @author Costin Leau
* @author David Turanski
*/
@@ -113,7 +114,7 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
AttributesFactory.validateAttributes(attributes);
final RegionFactory<K, V> regionFactory = (attributes != null ? c.createRegionFactory(attributes) : c
.<K, V> createRegionFactory());
.<K, V> createRegionFactory());
if (hubId != null) {
enableGateway = enableGateway == null ? true : enableGateway;
@@ -223,7 +224,7 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
* of this factory bean during the initialization process. The object is
* already initialized and configured by the factory bean before this method
* is invoked.
*
*
* @param attrFactory attribute factory
* @deprecated as of GemFire 6.5, the use of {@link AttributesFactory} has
* been deprecated
@@ -236,7 +237,7 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
* Post-process the region object for this factory bean during the
* initialization process. The object is already initialized and configured
* by the factory bean before this method is invoked.
*
*
* @param region
*/
protected void postProcess(Region<K, V> region) {
@@ -255,7 +256,7 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
// nothing to see folks, move on.
}
}
} if (destroy) {
region.destroyRegion();
region = null;
@@ -283,7 +284,7 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
* Sets the snapshots used for loading a newly <i>created</i> region. That
* is, the snapshot will be used <i>only</i> when a new region is created -
* if the region already exists, no loading will be performed.
*
*
* @see #setName(String)
* @param snapshot the snapshot to set
*/
@@ -295,7 +296,7 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
* Sets the cache listeners used for the region used by this factory. Used
* only when a new region is created.Overrides the settings specified
* through {@link #setAttributes(RegionAttributes)}.
*
*
* @param cacheListeners the cacheListeners to set on a newly created region
*/
public void setCacheListeners(CacheListener<K, V>[] cacheListeners) {
@@ -306,7 +307,7 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
* Sets the cache loader used for the region used by this factory. Used only
* when a new region is created.Overrides the settings specified through
* {@link #setAttributes(RegionAttributes)}.
*
*
* @param cacheLoader the cacheLoader to set on a newly created region
*/
public void setCacheLoader(CacheLoader<K, V> cacheLoader) {
@@ -317,7 +318,7 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
* Sets the cache writer used for the region used by this factory. Used only
* when a new region is created. Overrides the settings specified through
* {@link #setAttributes(RegionAttributes)}.
*
*
* @param cacheWriter the cacheWriter to set on a newly created region
*/
public void setCacheWriter(CacheWriter<K, V> cacheWriter) {
@@ -327,7 +328,7 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
/**
* Sets the region scope. Used only when a new region is created. Overrides
* the settings specified through {@link #setAttributes(RegionAttributes)}.
*
*
* @see Scope
* @param scope the region scope
*/
@@ -357,7 +358,7 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
}
/**
*
*
* @param gatewaySenders defined as Object for backward compatibility with
* Gemfire 6
*/
@@ -366,7 +367,7 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
}
/**
*
*
* @param asyncEventQueues defined as Object for backward compatibility with
* Gemfire 6
*/
@@ -386,7 +387,7 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
* Sets the region attributes used for the region used by this factory.
* Allows maximum control in specifying the region settings. Used only when
* a new region is created.
*
*
* @param attributes the attributes to set on a newly created region
*/
public void setAttributes(RegionAttributes<K, V> attributes) {
@@ -406,12 +407,14 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
*/
@Override
public void start() {
if (!ObjectUtils.isEmpty(gatewaySenders)) {
synchronized (gatewaySenders) {
for (Object obj : gatewaySenders) {
GatewaySender gws = (GatewaySender) obj;
System.out.println("manual start = " + gws.isManualStart());
if (!gws.isManualStart() && !gws.isRunning()) {
System.out.println("starting ...");
gws.start();
}
}

View File

@@ -18,6 +18,7 @@ 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;
@@ -32,7 +33,7 @@ import com.gemstone.gemfire.cache.wan.GatewayTransportFilter;
/**
* FactoryBean for creating a GemFire {@link GatewaySender}.
* @author David Turanski
*
*
*/
public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<GatewaySender>
implements SmartLifecycle {
@@ -75,7 +76,7 @@ implements SmartLifecycle {
private Integer socketReadTimeout;
/**
*
*
* @param cache the Gemfire cache
*/
public GatewaySenderFactoryBean(Cache cache) {
@@ -153,9 +154,9 @@ implements SmartLifecycle {
if (dispatcherThreads != null) {
gatewaySenderFactory.setDispatcherThreads(dispatcherThreads);
}
gatewaySenderFactory.setManualStart(true);
if (maximumQueueMemory != null) {
gatewaySenderFactory.setMaximumQueueMemory(maximumQueueMemory);
}
@@ -165,7 +166,9 @@ implements SmartLifecycle {
if (socketReadTimeout != null) {
gatewaySenderFactory.setSocketReadTimeout(socketReadTimeout);
}
gatewaySender = gatewaySenderFactory.create(getName(), remoteDistributedSystemId);
GatewaySenderWrapper wrapper = new GatewaySenderWrapper(gatewaySenderFactory.create(getName(), remoteDistributedSystemId));
wrapper.setManualStart(manualStart);
gatewaySender = wrapper;
}
public void setRemoteDistributedSystemId(int remoteDistributedSystemId) {

View File

@@ -0,0 +1,154 @@
package org.springframework.data.gemfire.wan;
import com.gemstone.gemfire.cache.util.Gateway;
import com.gemstone.gemfire.cache.wan.GatewayEventFilter;
import com.gemstone.gemfire.cache.wan.GatewaySender;
import com.gemstone.gemfire.cache.wan.GatewayTransportFilter;
import java.util.List;
/**
* Created by dturanski on 9/16/13.
*/
public class GatewaySenderWrapper implements GatewaySender {
private final GatewaySender delegate;
private boolean manualStart;
public GatewaySenderWrapper(GatewaySender delegate) {
this.delegate = delegate;
}
@Override
public void start() {
delegate.start();
}
@Override
public void stop() {
delegate.stop();
}
@Override
public void pause() {
delegate.pause();
}
@Override
public void resume() {
delegate.resume();
}
@Override
public boolean isRunning() {
return delegate.isRunning();
}
@Override
public boolean isPaused() {
return delegate.isPaused();
}
@Override
public void addGatewayEventFilter(GatewayEventFilter filter) {
delegate.addGatewayEventFilter(filter);
}
@Override
public void removeGatewayEventFilter(GatewayEventFilter filter) {
delegate.removeGatewayEventFilter(filter);
}
@Override
public String getId() {
return delegate.getId();
}
@Override
public int getRemoteDSId() {
return delegate.getRemoteDSId();
}
@Override
public int getSocketBufferSize() {
return delegate.getSocketBufferSize();
}
@Override
public int getSocketReadTimeout() {
return delegate.getSocketReadTimeout();
}
@Override
public String getDiskStoreName() {
return delegate.getDiskStoreName();
}
@Override
public int getMaximumQueueMemory() {
return delegate.getMaximumQueueMemory();
}
@Override
public int getBatchSize() {
return delegate.getBatchSize();
}
@Override
public int getBatchTimeInterval() {
return delegate.getBatchTimeInterval();
}
@Override
public boolean isBatchConflationEnabled() {
return delegate.isBatchConflationEnabled();
}
@Override
public boolean isPersistenceEnabled() {
return delegate.isPersistenceEnabled();
}
@Override
public int getAlertThreshold() {
return delegate.getAlertThreshold();
}
@Override
public List<GatewayEventFilter> getGatewayEventFilters() {
return delegate.getGatewayEventFilters();
}
@Override
public List<GatewayTransportFilter> getGatewayTransportFilters() {
return delegate.getGatewayTransportFilters();
}
@Override
public boolean isDiskSynchronous() {
return delegate.isDiskSynchronous();
}
@Override
public boolean isManualStart() {
return this.manualStart;
}
@Override
public boolean isParallel() {
return delegate.isParallel();
}
@Override
public int getDispatcherThreads() {
return delegate.getDispatcherThreads();
}
@Override
public Gateway.OrderPolicy getOrderPolicy() {
return delegate.getOrderPolicy();
}
public void setManualStart(boolean manualStart) {
this.manualStart = manualStart;
}
}

View File

@@ -48,9 +48,9 @@ import com.gemstone.gemfire.cache.wan.GatewayTransportFilter;
/**
* This test is only valid for GF 7.0 and above
*
*
* @author David Turanski
*
*
*/
public class GemfireV7GatewayNamespaceTest extends RecreatingContextTest {
@@ -148,7 +148,8 @@ public class GemfireV7GatewayNamespaceTest extends RecreatingContextTest {
assertTrue(transportFilters.get(0) instanceof TestTransportFilter);
assertEquals(1, gws.getRemoteDSId());
assertEquals(true, gws.isManualStart());
assertEquals(false, gws.isManualStart());
assertEquals(true,gws.isRunning());
assertEquals(10, gws.getAlertThreshold());
assertEquals(11, gws.getBatchSize());
assertEquals(3000, gws.getBatchTimeInterval());

View File

@@ -1,35 +1,41 @@
/*
* Copyright 2002-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.test;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.List;
import com.gemstone.gemfire.cache.Region;
import com.gemstone.gemfire.cache.util.Gateway.OrderPolicy;
import com.gemstone.gemfire.cache.wan.GatewayEventFilter;
import com.gemstone.gemfire.cache.wan.GatewaySender;
import com.gemstone.gemfire.cache.wan.GatewaySenderFactory;
import com.gemstone.gemfire.cache.wan.GatewayTransportFilter;
import com.sun.org.apache.xpath.internal.operations.Bool;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
/**
* @author David Turanski
*
*/
public class StubGatewaySenderFactory implements GatewaySenderFactory {
private int alertThreshold;
private boolean batchConflationEnabled;
private int batchSize;
@@ -49,6 +55,8 @@ public class StubGatewaySenderFactory implements GatewaySenderFactory {
private String name;
private boolean running = true;
private int remoteSystemId;
public StubGatewaySenderFactory() {
@@ -92,6 +100,18 @@ public class StubGatewaySenderFactory implements GatewaySenderFactory {
when(gatewaySender.isDiskSynchronous()).thenReturn(this.diskSynchronous);
when(gatewaySender.isParallel()).thenReturn(this.parallel);
when(gatewaySender.isPersistenceEnabled()).thenReturn(this.persistenceEnabled);
doAnswer(new Answer<Void>() {
public Void answer(InvocationOnMock invocation) {
running = true;
return null;
}
}).when(gatewaySender).start();
when(gatewaySender.isRunning()).thenAnswer(new Answer<Boolean>(){
@Override
public Boolean answer(InvocationOnMock invocation) throws Throwable {
return running;
}
});
return gatewaySender;
}

View File

@@ -7,17 +7,17 @@
xsi:schemaLocation="http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd" default-lazy-init="true">
<gfe:cache/>
<gfe:partitioned-region id="region-inner-gateway-sender" >
<gfe:gateway-sender
manual-start="true"
remote-distributed-system-id="1"
alert-threshold="10"
batch-size="11"
batch-time-interval="3000"
dispatcher-threads="2"
<gfe:gateway-sender
manual-start="false"
remote-distributed-system-id="1"
alert-threshold="10"
batch-size="11"
batch-time-interval="3000"
dispatcher-threads="2"
disk-store-ref="diskstore"
enable-batch-conflation="true"
maximum-queue-memory="50"
@@ -35,12 +35,12 @@
</gfe:gateway-sender>
<gfe:gateway-sender-ref bean="gateway-sender"/>
</gfe:partitioned-region>
<gfe:async-event-queue id="async-event-queue"
batch-size="10"
persistent="true"
disk-store-ref="diskstore"
maximum-queue-memory="50"
<gfe:async-event-queue id="async-event-queue"
batch-size="10"
persistent="true"
disk-store-ref="diskstore"
maximum-queue-memory="50"
parallel="true"
batch-conflation-enabled="true"
batch-time-interval="3"
@@ -51,21 +51,21 @@
<bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestAsyncEventListener"/>
</gfe:async-event-listener>
</gfe:async-event-queue>
<gfe:disk-store id="diskstore"/>
<gfe:gateway-receiver id="gateway-receiver"
<gfe:gateway-receiver id="gateway-receiver"
start-port="12345" end-port="23456" bind-address="192.168.0.1" maximum-time-between-pings="3000" socket-buffer-size="16536">
<gfe:transport-filter>
<bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestTransportFilter"/>
</gfe:transport-filter>
</gfe:gateway-receiver>
<!-- need manual-start=true for the unit test because GF will throw an exception if no locators are configured -->
<gfe:gateway-sender id="gateway-sender"
<gfe:gateway-sender id="gateway-sender"
remote-distributed-system-id="2"
alert-threshold="10"
batch-size="11"
alert-threshold="10"
batch-size="11"
dispatcher-threads="12"
manual-start="true">
<gfe:event-filter>
@@ -77,8 +77,8 @@
<bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestTransportFilter"/>
</gfe:transport-filter>
</gfe:gateway-sender>
<bean id="event-filter" class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestEventFilter"/>
<bean id="transport-filter" class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestTransportFilter"/>
</beans>