From a4b6acff075c124d6f6cd43e8d497ff4e77dac18 Mon Sep 17 00:00:00 2001 From: David Turanski Date: Mon, 16 Sep 2013 18:35:36 -0400 Subject: [PATCH] SGF-199 fixed GatewaySender manual-start --- .gitignore | 2 + .../data/gemfire/RegionFactoryBean.java | 33 ++-- .../gemfire/wan/GatewaySenderFactoryBean.java | 13 +- .../gemfire/wan/GatewaySenderWrapper.java | 154 ++++++++++++++++++ .../config/GemfireV7GatewayNamespaceTest.java | 7 +- .../test/StubGatewaySenderFactory.java | 28 +++- .../data/gemfire/config/gateway-v7-ns.xml | 48 +++--- 7 files changed, 234 insertions(+), 51 deletions(-) create mode 100644 src/main/java/org/springframework/data/gemfire/wan/GatewaySenderWrapper.java diff --git a/.gitignore b/.gitignore index fa419f58..a87cefbf 100644 --- a/.gitignore +++ b/.gitignore @@ -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 diff --git a/src/main/java/org/springframework/data/gemfire/RegionFactoryBean.java b/src/main/java/org/springframework/data/gemfire/RegionFactoryBean.java index 33fbcb76..a93f95d4 100644 --- a/src/main/java/org/springframework/data/gemfire/RegionFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/RegionFactoryBean.java @@ -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 extends RegionLookupFactoryBean imple AttributesFactory.validateAttributes(attributes); final RegionFactory regionFactory = (attributes != null ? c.createRegionFactory(attributes) : c - . createRegionFactory()); + . createRegionFactory()); if (hubId != null) { enableGateway = enableGateway == null ? true : enableGateway; @@ -223,7 +224,7 @@ public class RegionFactoryBean extends RegionLookupFactoryBean 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 extends RegionLookupFactoryBean 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 region) { @@ -255,7 +256,7 @@ public class RegionFactoryBean extends RegionLookupFactoryBean imple // nothing to see folks, move on. } } - + } if (destroy) { region.destroyRegion(); region = null; @@ -283,7 +284,7 @@ public class RegionFactoryBean extends RegionLookupFactoryBean imple * Sets the snapshots used for loading a newly created region. That * is, the snapshot will be used only 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 extends RegionLookupFactoryBean 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[] cacheListeners) { @@ -306,7 +307,7 @@ public class RegionFactoryBean extends RegionLookupFactoryBean 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 cacheLoader) { @@ -317,7 +318,7 @@ public class RegionFactoryBean extends RegionLookupFactoryBean 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 cacheWriter) { @@ -327,7 +328,7 @@ public class RegionFactoryBean extends RegionLookupFactoryBean 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 extends RegionLookupFactoryBean imple } /** - * + * * @param gatewaySenders defined as Object for backward compatibility with * Gemfire 6 */ @@ -366,7 +367,7 @@ public class RegionFactoryBean extends RegionLookupFactoryBean imple } /** - * + * * @param asyncEventQueues defined as Object for backward compatibility with * Gemfire 6 */ @@ -386,7 +387,7 @@ public class RegionFactoryBean extends RegionLookupFactoryBean 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 attributes) { @@ -406,12 +407,14 @@ public class RegionFactoryBean extends RegionLookupFactoryBean 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(); } } 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 af83f0f1..0ae7a81a 100644 --- a/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java @@ -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 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) { diff --git a/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderWrapper.java b/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderWrapper.java new file mode 100644 index 00000000..08a45688 --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderWrapper.java @@ -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 getGatewayEventFilters() { + return delegate.getGatewayEventFilters(); + } + + @Override + public List 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; + } +} diff --git a/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java b/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java index 84f53f1e..ea0260f0 100644 --- a/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java +++ b/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java @@ -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()); diff --git a/src/test/java/org/springframework/data/gemfire/test/StubGatewaySenderFactory.java b/src/test/java/org/springframework/data/gemfire/test/StubGatewaySenderFactory.java index ee27f4f6..9d542c55 100644 --- a/src/test/java/org/springframework/data/gemfire/test/StubGatewaySenderFactory.java +++ b/src/test/java/org/springframework/data/gemfire/test/StubGatewaySenderFactory.java @@ -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() { + public Void answer(InvocationOnMock invocation) { + running = true; + return null; + } + }).when(gatewaySender).start(); + when(gatewaySender.isRunning()).thenAnswer(new Answer(){ + @Override + public Boolean answer(InvocationOnMock invocation) throws Throwable { + return running; + } + }); return gatewaySender; } diff --git a/src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml b/src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml index 51dbaa6e..6c966fae 100644 --- a/src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml +++ b/src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml @@ -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"> - + - + - - - - + - - - + - @@ -77,8 +77,8 @@ - + - + \ No newline at end of file