From c66d4d240445b408574ec4f7c1c95863cab4ae41 Mon Sep 17 00:00:00 2001 From: David Turanski Date: Fri, 5 Oct 2012 07:24:49 -0400 Subject: [PATCH 1/2] updated build dependencies to gemfire 7 --- build.gradle | 1 + gradle.properties | 4 ++-- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/build.gradle b/build.gradle index ee183fd0..1a81ea91 100644 --- a/build.gradle +++ b/build.gradle @@ -14,6 +14,7 @@ group = 'org.springframework.data' repositories { maven { url "http://repo.springsource.org/libs-milestone" } maven { url "http://repo.springsource.org/plugins-release"} + maven { url "https://repo.springsource.org/ext-private-local"} } diff --git a/gradle.properties b/gradle.properties index fd0c89e8..2c7064cd 100644 --- a/gradle.properties +++ b/gradle.properties @@ -7,7 +7,7 @@ slf4jVersion = 1.6.4 # Common libraries springVersion = 3.1.2.RELEASE springDataCommonsVersion = 1.4.0.RC1 -gemfireVersion = 6.6.3 +gemfireVersion = 7.0.Beta-SNAPSHOT gemfireVersion6x = 6.6.3 # Testing @@ -20,7 +20,7 @@ cglibVersion = 2.2 ## OSGi ranges spring.range = "[3.0.0, 4.0.0)" -gemfire.range = "[6.5, 7.0)" +gemfire.range = "[7.0, 8.0)" # -------------------- # Project wide version From 154905b80843a4d5d2654d21010bfd4c02862eec Mon Sep 17 00:00:00 2001 From: David Turanski Date: Fri, 5 Oct 2012 10:43:41 -0400 Subject: [PATCH 2/2] initial 1.2.1 branch --- build.gradle | 2 + gradle.properties | 2 +- .../data/gemfire/CacheFactoryBean.java | 19 +- .../data/gemfire/RegionFactoryBean.java | 53 +- .../gemfire/config/AsyncEventQueueParser.java | 54 ++ .../gemfire/config/GatewayReceiverParser.java | 48 ++ .../gemfire/config/GatewaySenderParser.java | 67 +++ .../config/GemfireNamespaceHandler.java | 5 +- .../wan/AsyncEventQueueFactoryBean.java | 121 +++++ .../wan/GatewayReceiverFactoryBean.java | 123 +++++ .../gemfire/wan/GatewaySenderFactoryBean.java | 237 ++++++++ .../gemfire/config/spring-gemfire-1.2.xsd | 324 ++++++++++- .../gemfire/config/CacheNamespaceTest.java | 22 +- .../config/GemfireV7GatewayNamespaceTest.java | 507 ++++++++++++++++++ .../data/gemfire/config/cache-ns.xml | 8 +- .../data/gemfire/config/gateway-v7-ns.xml | 75 +++ 16 files changed, 1651 insertions(+), 16 deletions(-) create mode 100644 src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java create mode 100644 src/main/java/org/springframework/data/gemfire/config/GatewayReceiverParser.java create mode 100644 src/main/java/org/springframework/data/gemfire/config/GatewaySenderParser.java create mode 100644 src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java create mode 100644 src/main/java/org/springframework/data/gemfire/wan/GatewayReceiverFactoryBean.java create mode 100644 src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java create mode 100644 src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java create mode 100644 src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml diff --git a/build.gradle b/build.gradle index 1a81ea91..9a96e6d0 100644 --- a/build.gradle +++ b/build.gradle @@ -49,6 +49,8 @@ dependencies { // GemFire compile("com.gemstone.gemfire:gemfire:$gemfireVersion") + runtime("antlr:antlr:2.7.7") + //runtime("commons-modeler;commons-modeler:2.0.1") // Testing testCompile "junit:junit-dep:$junitVersion" diff --git a/gradle.properties b/gradle.properties index 2c7064cd..6e453d78 100644 --- a/gradle.properties +++ b/gradle.properties @@ -25,4 +25,4 @@ gemfire.range = "[7.0, 8.0)" # -------------------- # Project wide version # -------------------- -version=1.2.0.BUILD-SNAPSHOT +version=1.2.1.BUILD-SNAPSHOT diff --git a/src/main/java/org/springframework/data/gemfire/CacheFactoryBean.java b/src/main/java/org/springframework/data/gemfire/CacheFactoryBean.java index 957c4ef6..04002783 100644 --- a/src/main/java/org/springframework/data/gemfire/CacheFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/CacheFactoryBean.java @@ -48,6 +48,7 @@ import com.gemstone.gemfire.cache.DynamicRegionFactory; import com.gemstone.gemfire.cache.GemFireCache; import com.gemstone.gemfire.cache.TransactionListener; import com.gemstone.gemfire.cache.TransactionWriter; +import com.gemstone.gemfire.cache.util.GatewayConflictResolver; import com.gemstone.gemfire.distributed.DistributedMember; import com.gemstone.gemfire.distributed.DistributedSystem; import com.gemstone.gemfire.internal.cache.GemFireCacheImpl; @@ -130,6 +131,10 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl if (messageSyncInterval != null) { cacheImpl.setMessageSyncInterval(messageSyncInterval); } + + if (gatewayConflictResolver != null) { + cacheImpl.setGatewayConflictResolver((GatewayConflictResolver) gatewayConflictResolver); + } } } @@ -259,6 +264,9 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl protected List jndiDataSources; + // Defined this way for backward compatibility + protected Object gatewayConflictResolver; + @Override public void afterPropertiesSet() throws Exception { // initialize locator @@ -647,6 +655,15 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl public void setTransactionWriter(TransactionWriter transactionWriter) { this.transactionWriter = transactionWriter; } + /** + * Requires GemFire 7.0 or higher + * @param gatewayConflictResolver defined as Object in the signature for backward + * compatibility with Gemfire 6 compatibility. This must be an instance of + * {@link com.gemstone.gemfire.cache.util.GatewayConflictResolver} + */ + public void setGatewayConflictResolver(Object gatewayConflictResolver) { + this.gatewayConflictResolver = gatewayConflictResolver; + } /** * @@ -663,4 +680,4 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl public void setJndiDataSources(List jndiDataSources) { this.jndiDataSources = jndiDataSources; } -} \ No newline at end of file +} diff --git a/src/main/java/org/springframework/data/gemfire/RegionFactoryBean.java b/src/main/java/org/springframework/data/gemfire/RegionFactoryBean.java index 09460edf..ecceee72 100644 --- a/src/main/java/org/springframework/data/gemfire/RegionFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/RegionFactoryBean.java @@ -27,6 +27,7 @@ import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.ReflectionUtils; +import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueue; import com.gemstone.gemfire.cache.AttributesFactory; import com.gemstone.gemfire.cache.Cache; import com.gemstone.gemfire.cache.CacheClosedException; @@ -39,6 +40,7 @@ import com.gemstone.gemfire.cache.Region; import com.gemstone.gemfire.cache.RegionAttributes; import com.gemstone.gemfire.cache.RegionFactory; import com.gemstone.gemfire.cache.Scope; +import com.gemstone.gemfire.cache.wan.GatewaySender; /** * Base class for FactoryBeans used to create GemFire {@link Region}s. Will try @@ -69,6 +71,10 @@ public class RegionFactoryBean extends RegionLookupFactoryBean imple private CacheWriter cacheWriter; + private Object gatewaySenders[]; + + private Object asyncEventQueues[]; + private RegionAttributes attributes; private Scope scope; @@ -98,10 +104,9 @@ public class RegionFactoryBean extends RegionLookupFactoryBean imple Cache c = (Cache) cache; - if (attributes != null) { + if (attributes != null) AttributesFactory.validateAttributes(attributes); - } - + final RegionFactory regionFactory = (attributes != null ? c.createRegionFactory(attributes) : c . createRegionFactory()); @@ -111,17 +116,33 @@ public class RegionFactoryBean extends RegionLookupFactoryBean imple regionFactory.setGatewayHubId(hubId); } - - if (enableGateway !=null) { + if (enableGateway != null) { + if (enableGateway) { + Assert.notNull(hubId, "enableGateway requires the hubId property to be true"); + } regionFactory.setEnableGateway(enableGateway); } - if (!ObjectUtils.isEmpty(cacheListeners)) { for (CacheListener listener : cacheListeners) { regionFactory.addCacheListener(listener); } } + if (!ObjectUtils.isEmpty(gatewaySenders)) { + Assert.isTrue( + hubId == null, + "It is invalid to configure a region with both a hubId and gatewaySenders. Note that the enableGateway and hubId properties are deprecated since Gemfire 7.0"); + for (Object gatewaySender : gatewaySenders) { + regionFactory.addGatewaySenderId(((GatewaySender) gatewaySender).getId()); + } + } + + if (!ObjectUtils.isEmpty(asyncEventQueues)) { + for (Object asyncEventQueue : asyncEventQueues) { + regionFactory.addAsyncEventQueueId(((AsyncEventQueue) asyncEventQueue).getId()); + } + } + if (cacheLoader != null) { regionFactory.setCacheLoader(cacheLoader); } @@ -349,6 +370,24 @@ public class RegionFactoryBean extends RegionLookupFactoryBean imple this.diskStoreName = diskStoreName; } + /** + * + * @param gatewaySenders defined as Object for backward compatibility with + * Gemfire 6 + */ + public void setGatewaySenders(Object[] gatewaySenders) { + this.gatewaySenders = gatewaySenders; + } + + /** + * + * @param asyncEventQueues defined as Object for backward compatibility with + * Gemfire 6 + */ + public void setAsyncEventQueues(Object[] asyncEventQueues) { + this.asyncEventQueues = asyncEventQueues; + } + public void setEnableGateway(boolean enableGateway) { this.enableGateway = enableGateway; } @@ -375,4 +414,4 @@ public class RegionFactoryBean extends RegionLookupFactoryBean imple protected boolean isNotPersistent() { return persistent != null && !persistent; } -} \ No newline at end of file +} diff --git a/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java b/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java new file mode 100644 index 00000000..ef7d697c --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java @@ -0,0 +1,54 @@ +/* + * Copyright 2010-2012 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.config; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.data.gemfire.wan.AsyncEventQueueFactoryBean; +import org.springframework.util.xml.DomUtils; +import org.w3c.dom.Element; + +import com.gemstone.gemfire.internal.lang.StringUtils; + +/** + * @author David Turanski + * + */ +public class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser { + + @Override + protected Class getBeanClass(Element element) { + return AsyncEventQueueFactoryBean.class; + } + + @Override + protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + builder.setLazyInit(false); + Element asyncEventListenerElement = DomUtils.getChildElementByTagName(element, "async-event-listener"); + Object asyncEventListener = ParsingUtils.parseRefOrSingleNestedBeanDeclaration(parserContext, + asyncEventListenerElement, builder); + String cacheName = StringUtils.isEmpty(element.getAttribute("cache-ref")) ? "gemfireCache" : element + .getAttribute("cache-ref"); + builder.addConstructorArgReference(cacheName); + builder.addConstructorArgValue(asyncEventListener); + ParsingUtils.setPropertyValue(element, builder, "batch-size"); + ParsingUtils.setPropertyValue(element, builder, "maximum-queue-memory"); + ParsingUtils.setPropertyValue(element, builder, "disk-store-ref"); + ParsingUtils.setPropertyValue(element, builder, "persistent"); + ParsingUtils.setPropertyValue(element, builder, "parallel"); + } +} diff --git a/src/main/java/org/springframework/data/gemfire/config/GatewayReceiverParser.java b/src/main/java/org/springframework/data/gemfire/config/GatewayReceiverParser.java new file mode 100644 index 00000000..58a10827 --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/config/GatewayReceiverParser.java @@ -0,0 +1,48 @@ +/* + * Copyright 2010-2012 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.config; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.data.gemfire.wan.GatewayReceiverFactoryBean; +import org.springframework.util.StringUtils; +import org.w3c.dom.Element; + +/** + * @author David Turanski + * + */ +class GatewayReceiverParser extends AbstractSimpleBeanDefinitionParser { + @Override + protected Class getBeanClass(Element element) { + return GatewayReceiverFactoryBean.class; + } + + @Override + protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + builder.setLazyInit(false); + String cacheRef = element.getAttribute("cache-ref"); + // add cache reference (fallback to default if nothing is specified) + builder.addConstructorArgReference((StringUtils.hasText(cacheRef) ? cacheRef : "gemfireCache")); + ParsingUtils.setPropertyValue(element, builder, "start-port"); + ParsingUtils.setPropertyValue(element, builder, "end-port"); + ParsingUtils.setPropertyValue(element, builder, "socket-buffer-size"); + ParsingUtils.setPropertyValue(element, builder, "maximum-time-between-pings"); + ParsingUtils.setPropertyValue(element, builder, "bind-address"); + ParsingUtils.parseTransportFilters(element, parserContext, builder); + } +} diff --git a/src/main/java/org/springframework/data/gemfire/config/GatewaySenderParser.java b/src/main/java/org/springframework/data/gemfire/config/GatewaySenderParser.java new file mode 100644 index 00000000..a42392b0 --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/config/GatewaySenderParser.java @@ -0,0 +1,67 @@ +/* + * Copyright 2010-2012 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.config; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.data.gemfire.wan.GatewaySenderFactoryBean; +import org.springframework.util.StringUtils; +import org.springframework.util.xml.DomUtils; +import org.w3c.dom.Element; + +/** + * @author David Turanski + * + */ +class GatewaySenderParser extends AbstractSimpleBeanDefinitionParser { + + @Override + protected Class getBeanClass(Element element) { + return GatewaySenderFactoryBean.class; + } + + @Override + protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + builder.setLazyInit(false); + String cacheRef = element.getAttribute("cache-ref"); + // add cache reference (fallback to default if nothing is specified) + builder.addConstructorArgReference((StringUtils.hasText(cacheRef) ? cacheRef : "gemfireCache")); + ParsingUtils.setPropertyValue(element, builder, "alert-threshold"); + ParsingUtils.setPropertyValue(element, builder, "batch-size"); + ParsingUtils.setPropertyValue(element, builder, "batch-time-interval"); + ParsingUtils.setPropertyValue(element, builder, "disk-store-ref"); + ParsingUtils.setPropertyValue(element, builder, "disk-synchronous"); + ParsingUtils.setPropertyValue(element, builder, "dispatcher-threads"); + ParsingUtils.setPropertyValue(element, builder, "enable-batch-conflation"); + ParsingUtils.setPropertyValue(element, builder, "manual-start"); + ParsingUtils.setPropertyValue(element, builder, "maximum-queue-memory"); + ParsingUtils.setPropertyValue(element, builder, "order-policy"); + ParsingUtils.setPropertyValue(element, builder, "remote-distributed-system-id"); + ParsingUtils.setPropertyValue(element, builder, "socket-buffer-size"); + ParsingUtils.setPropertyValue(element, builder, "socket-read-timeout"); + ParsingUtils.setPropertyValue(element, builder, "persistent"); + ParsingUtils.setPropertyValue(element, builder, "parallel"); + + Element eventFilterElement = DomUtils.getChildElementByTagName(element, "event-filter"); + if (eventFilterElement != null) { + builder.addPropertyValue("eventFilters", + ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, eventFilterElement, builder)); + } + + ParsingUtils.parseTransportFilters(element, parserContext, builder); + } +} diff --git a/src/main/java/org/springframework/data/gemfire/config/GemfireNamespaceHandler.java b/src/main/java/org/springframework/data/gemfire/config/GemfireNamespaceHandler.java index 13bd5206..191aa566 100644 --- a/src/main/java/org/springframework/data/gemfire/config/GemfireNamespaceHandler.java +++ b/src/main/java/org/springframework/data/gemfire/config/GemfireNamespaceHandler.java @@ -61,8 +61,11 @@ class GemfireNamespaceHandler extends NamespaceHandlerSupport { registerBeanDefinitionParser("cache-server", new CacheServerParser()); registerBeanDefinitionParser("transaction-manager", new TransactionManagerParser()); registerBeanDefinitionParser("cq-listener-container", new GemfireListenerContainerParser()); + registerBeanDefinitionParser("async-event-queue", new AsyncEventQueueParser()); + registerBeanDefinitionParser("gateway-sender", new GatewaySenderParser()); + registerBeanDefinitionParser("gateway-receiver", new GatewayReceiverParser()); registerBeanDefinitionParser("function-service", new FunctionServiceParser()); // V6 WAN parsers registerBeanDefinitionParser("gateway-hub", new GatewayHubParser()); } -} \ No newline at end of file +} diff --git a/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java b/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java new file mode 100644 index 00000000..255b2381 --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java @@ -0,0 +1,121 @@ +/* + * Copyright 2010-2012 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 org.springframework.util.Assert; + +import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueue; +import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueueFactory; +import com.gemstone.gemfire.cache.Cache; +import com.gemstone.gemfire.cache.CacheClosedException; +import com.gemstone.gemfire.cache.asyncqueue.AsyncEventListener; + +/** + * FactoryBean for creating GemFire {@link AsyncEventQueue}s. + * @author David Turanski + * + */ +public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean { + + public void setDiskStoreRef(String diskStoreRef) { + this.diskStoreRef = diskStoreRef; + } + + private final AsyncEventListener asyncEventListener; + + private AsyncEventQueue asyncEventQueue; + + private Integer batchSize; + + private Integer maximumQueueMemory; + + private Boolean persistent; + + private String diskStoreRef; + + /** + * + * @param cache the gemfire cache + * @param asyncEventListener required {@link AsyncEventListener} + */ + public AsyncEventQueueFactoryBean(Cache cache, AsyncEventListener asyncEventListener) { + super(cache); + this.asyncEventListener = asyncEventListener; + } + + @Override + public AsyncEventQueue getObject() throws Exception { + return asyncEventQueue; + } + + @Override + public Class getObjectType() { + return AsyncEventQueue.class; + } + + @Override + protected void doInit() { + Assert.notNull(asyncEventListener, "AsyncEventListener cannot be null"); + AsyncEventQueueFactory asyncEventQueueFactory = null; + if (this.factory == null) { + asyncEventQueueFactory = cache.createAsyncEventQueueFactory(); + } else { + asyncEventQueueFactory = (AsyncEventQueueFactory)factory; + } + + if (persistent != null) { + asyncEventQueueFactory.setPersistent(persistent); + } + if (batchSize != null) { + asyncEventQueueFactory.setBatchSize(batchSize); + } + if (diskStoreRef != null) { + persistent = (persistent == null) ? Boolean.TRUE : persistent; + Assert.isTrue(persistent, "specifying a disk store requires persistent property to be true"); + asyncEventQueueFactory.setDiskStoreName(diskStoreRef); + } + + if (maximumQueueMemory != null) { + asyncEventQueueFactory.setMaximumQueueMemory(maximumQueueMemory); + } + + asyncEventQueue = asyncEventQueueFactory.create(name, asyncEventListener); + } + + @Override + public void destroy() throws Exception { + if (!cache.isClosed()) { + try { + asyncEventListener.close(); + } + catch (CacheClosedException cce) { + // nothing to see folks, move on. + } + } + } + + public void setBatchSize(Integer batchSize) { + this.batchSize = batchSize; + } + + public void setMaximumQueueMemory(Integer maximumQueueMemory) { + this.maximumQueueMemory = maximumQueueMemory; + } + + public void setPersistent(Boolean persistent) { + this.persistent = persistent; + } +} diff --git a/src/main/java/org/springframework/data/gemfire/wan/GatewayReceiverFactoryBean.java b/src/main/java/org/springframework/data/gemfire/wan/GatewayReceiverFactoryBean.java new file mode 100644 index 00000000..81137be1 --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/wan/GatewayReceiverFactoryBean.java @@ -0,0 +1,123 @@ +/* + * Copyright 2010-2012 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 java.util.List; + +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; + +import com.gemstone.gemfire.cache.Cache; +import com.gemstone.gemfire.cache.wan.GatewayReceiver; +import com.gemstone.gemfire.cache.wan.GatewayReceiverFactory; +import com.gemstone.gemfire.cache.wan.GatewayTransportFilter; + +/** + * FactoryBean for creating a GemFire {@link GatewayReceiver}. + * + * @author David Turanski + * + */ +public class GatewayReceiverFactoryBean extends AbstractWANComponentFactoryBean { + private GatewayReceiver gatewayReceiver; + + private List transportFilters; + + private Integer startPort; + + private Integer endPort; + + private Integer maximumTimeBetweenPings; + + private Integer socketBufferSize; + + private String bindAddress; + + /** + * + * @param cache + */ + public GatewayReceiverFactoryBean(Cache cache) { + super(cache); + // Bean name not required. + this.name = "gateway-receiver"; + } + + @Override + public GatewayReceiver getObject() throws Exception { + return gatewayReceiver; + } + + @Override + public Class getObjectType() { + return GatewayReceiver.class; + } + + @Override + protected void doInit() { + GatewayReceiverFactory gatewayReceiverFactory = cache.createGatewayReceiverFactory(); + if (!CollectionUtils.isEmpty(transportFilters)) { + for (GatewayTransportFilter transportFilter : transportFilters) { + gatewayReceiverFactory.addGatewayTransportFilter(transportFilter); + } + } + int minPort = (startPort == null) ? GatewayReceiver.DEFAULT_START_PORT : startPort; + int maxPort = (endPort == null) ? GatewayReceiver.DEFAULT_END_PORT : endPort; + Assert.isTrue(minPort <= maxPort, "startPort must be less then or equal to " + maxPort); + gatewayReceiverFactory.setStartPort(minPort); + gatewayReceiverFactory.setEndPort(maxPort); + if (socketBufferSize != null) { + gatewayReceiverFactory.setSocketBufferSize(socketBufferSize); + } + if (maximumTimeBetweenPings != null) { + gatewayReceiverFactory.setMaximumTimeBetweenPings(maximumTimeBetweenPings); + } + if (bindAddress != null) { + gatewayReceiverFactory.setBindAddress(bindAddress); + } + + gatewayReceiver = gatewayReceiverFactory.create(); + } + + public void setGatewayReceiver(GatewayReceiver gatewayReceiver) { + this.gatewayReceiver = gatewayReceiver; + } + + public void setTransportFilters(List transportFilters) { + this.transportFilters = transportFilters; + } + + public void setStartPort(Integer startPort) { + this.startPort = startPort; + } + + public void setEndPort(Integer endPort) { + this.endPort = endPort; + } + + public void setMaximumTimeBetweenPings(Integer maximumTimeBetweenPings) { + this.maximumTimeBetweenPings = maximumTimeBetweenPings; + } + + public void setSocketBufferSize(Integer socketBufferSize) { + this.socketBufferSize = socketBufferSize; + } + + public void setBindAddress(String bindAddress) { + this.bindAddress = bindAddress; + } + +} diff --git a/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java b/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java new file mode 100644 index 00000000..829ef50d --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java @@ -0,0 +1,237 @@ +/* + * Copyright 2010-2012 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 java.util.Arrays; +import java.util.List; + +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; + +import com.gemstone.gemfire.cache.Cache; +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.GatewaySenderFactory; +import com.gemstone.gemfire.cache.wan.GatewayTransportFilter; + +/** + * FactoryBean for creating a GemFire {@link GatewaySender}. + * @author David Turanski + * + */ +public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean { + private static List validOrderPolicyValues = Arrays.asList("KEY", "PARTITION", "THREAD"); + + private GatewaySender gatewaySender; + + private int remoteDistributedSystemId; + + 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; + + private Integer maximumQueueMemory; + + private String orderPolicy; + + private Boolean parallel; + + private Boolean persistent; + + private Integer socketBufferSize; + + private Integer socketReadTimeout; + + /** + * + * @param cache the Gemfire cache + */ + public GatewaySenderFactoryBean(Cache cache) { + super(cache); + } + + @Override + public GatewaySender getObject() throws Exception { + return gatewaySender; + } + + @Override + public Class getObjectType() { + return GatewaySender.class; + } + + @Override + protected void doInit() { + GatewaySenderFactory gatewaySenderFactory = null; + if (this.factory == null) { + gatewaySenderFactory = cache.createGatewaySenderFactory(); + } else { + gatewaySenderFactory = (GatewaySenderFactory)factory; + } + if (diskStoreRef != null) { + persistent = (persistent == null) ? Boolean.TRUE : persistent; + Assert.isTrue(persistent, "specifying a disk store requires 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"); + gatewaySenderFactory.setDiskSynchronous(diskSynchronous); + } + + if (persistent != null) { + gatewaySenderFactory.setPersistenceEnabled(persistent); + } + + parallel = (parallel == null) ? Boolean.FALSE : parallel; + + gatewaySenderFactory.setParallel(parallel); + + if (orderPolicy != null) { + Assert.isTrue(parallel, "specifying an order policy requires the parallel property to be true"); + + Assert.isTrue(validOrderPolicyValues.contains(orderPolicy.toUpperCase()), "The value of order policy:'" + + orderPolicy + "' is invalid"); + gatewaySenderFactory.setOrderPolicy(Gateway.OrderPolicy.valueOf(orderPolicy.toUpperCase())); + } + + if (!CollectionUtils.isEmpty(eventFilters)) { + for (GatewayEventFilter eventFilter : eventFilters) { + gatewaySenderFactory.addGatewayEventFilter(eventFilter); + } + } + if (!CollectionUtils.isEmpty(transportFilters)) { + for (GatewayTransportFilter transportFilter : transportFilters) { + gatewaySenderFactory.addGatewayTransportFilter(transportFilter); + } + } + if (alertThreshold != null) { + gatewaySenderFactory.setAlertThreshold(alertThreshold); + } + if (enableBatchConflation != null) { + gatewaySenderFactory.setBatchConflationEnabled(enableBatchConflation); + } + if (batchSize != null) { + gatewaySenderFactory.setBatchSize(batchSize); + } + if (batchTimeInterval != null) { + gatewaySenderFactory.setBatchTimeInterval(batchTimeInterval); + } + if (dispatcherThreads != null) { + gatewaySenderFactory.setDispatcherThreads(dispatcherThreads); + } + if (manualStart != null) { + gatewaySenderFactory.setManualStart(manualStart); + } + if (maximumQueueMemory != null) { + gatewaySenderFactory.setMaximumQueueMemory(maximumQueueMemory); + } + if (socketBufferSize != null) { + gatewaySenderFactory.setSocketBufferSize(socketBufferSize); + } + if (socketReadTimeout != null) { + gatewaySenderFactory.setSocketReadTimeout(socketReadTimeout); + } + gatewaySender = gatewaySenderFactory.create(name, remoteDistributedSystemId); + } + + public void setRemoteDistributedSystemId(int remoteDistributedSystemId) { + this.remoteDistributedSystemId = remoteDistributedSystemId; + } + + public void setEventFilters(List gatewayEventFilters) { + this.eventFilters = gatewayEventFilters; + } + + public void setTransportFilters(List gatewayTransportFilters) { + this.transportFilters = gatewayTransportFilters; + } + + public void setAlertThreshold(Integer alertThreshold) { + this.alertThreshold = alertThreshold; + } + + public void setEnableBatchConflation(Boolean enableBatchConflation) { + this.enableBatchConflation = enableBatchConflation; + } + + public void setBatchSize(Integer batchSize) { + this.batchSize = batchSize; + } + + public void setBatchTimeInterval(Integer batchTimeInterval) { + this.batchTimeInterval = batchTimeInterval; + } + + public void setDiskStoreRef(String diskStoreRef) { + this.diskStoreRef = diskStoreRef; + } + + public void setDiskSynchronous(Boolean diskSynchronous) { + this.diskSynchronous = diskSynchronous; + } + + public void setDispatcherThreads(Integer dispatcherThreads) { + this.dispatcherThreads = dispatcherThreads; + } + + public void setManualStart(Boolean manualStart) { + this.manualStart = manualStart; + } + + public void setMaximumQueueMemory(Integer maximumQueueMemory) { + this.maximumQueueMemory = maximumQueueMemory; + } + + public void setOrderPolicy(String orderPolicy) { + this.orderPolicy = orderPolicy; + } + + public void setParallel(Boolean parallel) { + this.parallel = parallel; + } + + public void setPersistent(Boolean persistent) { + this.persistent = persistent; + } + + public void setSocketBufferSize(Integer socketBufferSize) { + this.socketBufferSize = socketBufferSize; + } + + public void setSocketReadTimeout(Integer socketReadTimeout) { + this.socketReadTimeout = socketReadTimeout; + } + +} diff --git a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.2.xsd b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.2.xsd index 2de2c78c..283b747e 100755 --- a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.2.xsd +++ b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.2.xsd @@ -35,7 +35,45 @@ and may be nested or referenced. ]]> - + + + + + + + + + + + + + + + + + + + + + + + + + @@ -468,6 +506,40 @@ when one or more missing required roles is restored to the distributed membershi + + + + + + + + + + + + + + + + + + + + + + + + @@ -591,7 +663,7 @@ The fully qualified class name of the expected value type use="optional"> @@ -599,7 +671,7 @@ The fully qualified class name of the expected value type use="optional"> @@ -2130,6 +2202,137 @@ Specifies a data type if other than java.lang.String + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + @@ -2342,6 +2652,7 @@ use inner bean declarations. @@ -2349,6 +2660,7 @@ use inner bean declarations. minOccurs="0" maxOccurs="unbounded"> @@ -2412,6 +2724,7 @@ Specifies the startup policy (primary,secondary, none) for the gateway hub @@ -2426,6 +2739,7 @@ Specifies the startup policy (primary,secondary, none) for the gateway hub @@ -2585,8 +2900,9 @@ Specifies the number of parallel threads - \ No newline at end of file + diff --git a/src/test/java/org/springframework/data/gemfire/config/CacheNamespaceTest.java b/src/test/java/org/springframework/data/gemfire/config/CacheNamespaceTest.java index ae9e67e1..56f23d71 100644 --- a/src/test/java/org/springframework/data/gemfire/config/CacheNamespaceTest.java +++ b/src/test/java/org/springframework/data/gemfire/config/CacheNamespaceTest.java @@ -33,6 +33,11 @@ import org.springframework.data.gemfire.TestUtils; import org.springframework.data.gemfire.client.ClientCacheFactoryBean; import org.springframework.test.util.ReflectionTestUtils; +import com.gemstone.gemfire.cache.Cache; +import com.gemstone.gemfire.cache.util.GatewayConflictHelper; +import com.gemstone.gemfire.cache.util.GatewayConflictResolver; +import com.gemstone.gemfire.cache.util.TimestampedEntryEvent; + /** * @author Costin Leau */ @@ -49,6 +54,7 @@ public class CacheNamespaceTest extends RecreatingContextTest { testNamedCache(); testCacheWithXml(); testHeapTunedCache(); + testCacheWithGatewayConflictResolver(); } private void testBasicCache() throws Exception { @@ -80,6 +86,12 @@ public class CacheNamespaceTest extends RecreatingContextTest { } + private void testCacheWithGatewayConflictResolver() { + Cache cache = ctx.getBean("cache-with-conflict-resolver", Cache.class); + assertNotNull(cache.getGatewayConflictResolver()); + assertTrue(cache.getGatewayConflictResolver() instanceof TestConflictResolver); + } + @Test(expected = IllegalArgumentException.class) public void testNoBeanFactory() throws Exception { assertTrue(ctx.containsBean("no-bl")); @@ -122,4 +134,12 @@ public class CacheNamespaceTest extends RecreatingContextTest { assertEquals(70, chp, 0.0001); assertEquals(60, ehp, 0.0001); } -} \ No newline at end of file + + public static class TestConflictResolver implements GatewayConflictResolver { + @Override + public void onEvent(TimestampedEntryEvent arg0, GatewayConflictHelper arg1) { + // TODO Auto-generated method stub + + } + } +} diff --git a/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java b/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java new file mode 100644 index 00000000..c5134ece --- /dev/null +++ b/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java @@ -0,0 +1,507 @@ +/* + * Copyright 2010-2012 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.config; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.File; +import java.io.FilenameFilter; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.ArrayList; +import java.util.List; + +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.Test; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.data.gemfire.RecreatingContextTest; +import org.springframework.data.gemfire.RegionFactoryBean; +import org.springframework.data.gemfire.TestUtils; +import org.springframework.data.gemfire.wan.AsyncEventQueueFactoryBean; +import org.springframework.data.gemfire.wan.GatewaySenderFactoryBean; + +import com.gemstone.gemfire.cache.Cache; +import com.gemstone.gemfire.cache.Region; +import com.gemstone.gemfire.cache.asyncqueue.AsyncEvent; +import com.gemstone.gemfire.cache.asyncqueue.AsyncEventListener; +import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueue; +import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueueFactory; +import com.gemstone.gemfire.cache.util.Gateway.OrderPolicy; +import com.gemstone.gemfire.cache.wan.GatewayEventFilter; +import com.gemstone.gemfire.cache.wan.GatewayQueueEvent; +import com.gemstone.gemfire.cache.wan.GatewayReceiver; +import com.gemstone.gemfire.cache.wan.GatewaySender; +import com.gemstone.gemfire.cache.wan.GatewaySenderFactory; +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 implements BeanPostProcessor { + + @Override + protected String location() { + return "/org/springframework/data/gemfire/config/gateway-v7-ns.xml"; + } + + @Override + protected void configureContext() { + ctx.getBeanFactory().addBeanPostProcessor(this); + } + + @Before + @Override + public void createCtx() { + if (ParsingUtils.GEMFIRE_VERSION.startsWith("7")) { + super.createCtx(); + } + } + + /* + * Faster this way + */ + @Test + public void test() throws Exception { + if (ctx != null) { + testGatewaySender(); + testInnerGatewaySender(); + testInnerGatewayReceiver(); + testAsyncEventQueue(); + } + } + + @AfterClass + public static void tearDown() { + for (String name : new File(".").list(new FilenameFilter() { + + @Override + public boolean accept(File dir, String name) { + return name.startsWith("BACKUP"); + } + })) { + new File(name).delete(); + } + } + + /** + * + */ + private void testAsyncEventQueue() { + AsyncEventQueue aseq = ctx.getBean("async-event-queue", AsyncEventQueue.class); + assertEquals(10, aseq.getBatchSize()); + assertTrue(aseq.isPersistent()); + assertEquals("diskstore", aseq.getDiskStoreName()); + assertEquals(50, aseq.getMaximumQueueMemory()); + } + + private void testGatewaySender() throws Exception { + GatewaySenderFactoryBean gwsfb = ctx.getBean("&gateway-sender", GatewaySenderFactoryBean.class); + Cache cache = TestUtils.readField("cache", gwsfb); + assertNotNull(cache); + List eventFilters = TestUtils.readField("eventFilters", gwsfb); + assertNotNull(eventFilters); + assertEquals(2, eventFilters.size()); + assertTrue(eventFilters.get(0) instanceof TestEventFilter); + + List transportFilters = TestUtils.readField("transportFilters", gwsfb); + assertNotNull(transportFilters); + assertEquals(2, transportFilters.size()); + assertTrue(transportFilters.get(0) instanceof TestTransportFilter); + + assertEquals(2, TestUtils.readField("remoteDistributedSystemId", gwsfb)); + assertEquals(10, TestUtils.readField("alertThreshold", gwsfb)); + assertEquals(11, TestUtils.readField("batchSize", gwsfb)); + assertEquals(12, TestUtils.readField("dispatcherThreads", gwsfb)); + assertEquals(true, TestUtils.readField("manualStart", gwsfb)); + } + + @SuppressWarnings("rawtypes") + private void testInnerGatewaySender() throws Exception { + Region region = ctx.getBean("region-inner-gateway-sender", Region.class); + GatewaySender gws = ctx.getBean("gateway-sender", GatewaySender.class); + assertNotNull(region.getAttributes().getGatewaySenderIds()); + assertEquals(2, region.getAttributes().getGatewaySenderIds().size()); + +// // Isolate the inner gateway +// Set gatewaySenders = region.getAttributes().getGatewaySenderIds(); +// assertTrue(gatewaySenders.remove(gws)); +// gatewaySenders.remove(gws); + + RegionFactoryBean rfb = ctx.getBean("®ion-inner-gateway-sender", RegionFactoryBean.class); + Object[] gwsenders = TestUtils.readField("gatewaySenders", rfb); + gws = (GatewaySender)gwsenders[0]; + List eventFilters = gws.getGatewayEventFilters(); + assertNotNull(eventFilters); + assertEquals(1, eventFilters.size()); + assertTrue(eventFilters.get(0) instanceof TestEventFilter); + + List transportFilters = gws.getGatewayTransportFilters(); + + assertNotNull(transportFilters); + assertEquals(1, transportFilters.size()); + assertTrue(transportFilters.get(0) instanceof TestTransportFilter); + + assertEquals(1, gws.getRemoteDSId()); + assertEquals(true, gws.isManualStart()); + assertEquals(10, gws.getAlertThreshold()); + assertEquals(11, gws.getBatchSize()); + assertEquals(3000, gws.getBatchTimeInterval()); + assertEquals(2, gws.getDispatcherThreads()); + assertEquals("diskstore", gws.getDiskStoreName()); + assertTrue(gws.isBatchConflationEnabled()); + assertEquals(50, gws.getMaximumQueueMemory()); + assertEquals(OrderPolicy.THREAD, gws.getOrderPolicy()); + assertTrue(gws.isPersistenceEnabled()); + assertTrue(gws.isParallel()); + assertEquals(16536, gws.getSocketBufferSize()); + assertEquals(3000, gws.getSocketReadTimeout()); + } + + private void testInnerGatewayReceiver() { + GatewayReceiver gwr = ctx.getBean("gateway-receiver", GatewayReceiver.class); + assertEquals(12345, gwr.getStartPort()); + assertEquals(23456, gwr.getEndPort()); + assertEquals("192.168.0.1", gwr.getBindAddress()); + assertEquals(3000, gwr.getMaximumTimeBetweenPings()); + assertEquals(16536, gwr.getSocketBufferSize()); + } + + @SuppressWarnings("rawtypes") + public static class TestEventFilter implements GatewayEventFilter { + + @Override + public void close() { + // TODO Auto-generated method stub + } + + @Override + public void afterAcknowledgement(GatewayQueueEvent arg0) { + // TODO Auto-generated method stub + + } + + @Override + public boolean beforeEnqueue(GatewayQueueEvent arg0) { + // TODO Auto-generated method stub + return false; + } + + @Override + public boolean beforeTransmit(GatewayQueueEvent arg0) { + // TODO Auto-generated method stub + return false; + } + + } + + public static class TestTransportFilter implements GatewayTransportFilter { + + @Override + public void close() { + // TODO Auto-generated method stub + + } + + @Override + public InputStream getInputStream(InputStream arg0) { + // TODO Auto-generated method stub + return null; + } + + @Override + public OutputStream getOutputStream(OutputStream arg0) { + // TODO Auto-generated method stub + return null; + } + + } + + @SuppressWarnings("rawtypes") + public static class TestAsyncEventListener implements AsyncEventListener { + + @Override + public void close() { + // TODO Auto-generated method stub + } + + @Override + public boolean processEvents(List arg0) { + // TODO Auto-generated method stub + return false; + } + + } + + + public static class StubAsyncEventQueueFactory implements AsyncEventQueueFactory { + + private AsyncEventListener listener; + + private AsyncEventQueue asyncEventQueue = mock(AsyncEventQueue.class); + + private boolean persistent; + private int maxQueueMemory; + private String diskStoreName; + private int batchSize; + + private String name; + + @Override + public AsyncEventQueue create(String name, AsyncEventListener listener) { + this.name = name; + this.listener = listener; + + when(asyncEventQueue.getAsyncEventListener()).thenReturn(this.listener); + when(asyncEventQueue.getBatchSize()).thenReturn(this.batchSize); + when(asyncEventQueue.getDiskStoreName()).thenReturn(this.diskStoreName); + when(asyncEventQueue.isPersistent()).thenReturn(this.persistent); + when(asyncEventQueue.getId()).thenReturn(this.name); + when(asyncEventQueue.getMaximumQueueMemory()).thenReturn(this.maxQueueMemory); + return this.asyncEventQueue; + } + + @Override + public AsyncEventQueueFactory setBatchSize(int batchSize) { + this.batchSize = batchSize; + return this; + } + + @Override + public AsyncEventQueueFactory setDiskStoreName(String diskStoreName) { + this.diskStoreName = diskStoreName; + return this; + } + + @Override + public AsyncEventQueueFactory setMaximumQueueMemory(int maxQueueMemory) { + this.maxQueueMemory = maxQueueMemory; + return this; + } + + @Override + public AsyncEventQueueFactory setPersistent(boolean persistent) { + this.persistent = persistent; + return this; + } + + @Override + public AsyncEventQueueFactory setBatchTimeInterval(int arg0) { + // TODO Auto-generated method stub + return null; + } + + @Override + public AsyncEventQueueFactory setParallel(boolean arg0) { + // TODO Auto-generated method stub + return null; + } + } + + public static class StubGWSenderFactory implements GatewaySenderFactory { + + private GatewaySender gatewaySender = mock(GatewaySender.class); + + private int alertThreshold; + private boolean batchConflationEnabled; + private int batchSize; + private int batchTimeInterval; + private String diskStoreName; + private boolean diskSynchronous; + private int dispatcherThreads; + private boolean manualStart; + private int maxQueueMemory; + private OrderPolicy orderPolicy; + private boolean parallel; + private boolean persistenceEnabled; + private int socketBufferSize; + private int socketReadTimeout; + private List eventFilters; + private List transportFilters; + + private String name; + + private int remoteSystemId; + + public StubGWSenderFactory() { + this.eventFilters = new ArrayList(); + this.transportFilters = new ArrayList(); + + } + + @Override + public GatewaySenderFactory addGatewayEventFilter(GatewayEventFilter filter) { + eventFilters.add(filter); + return this; + } + + @Override + public GatewaySenderFactory addGatewayTransportFilter(GatewayTransportFilter filter) { + transportFilters.add(filter); + return this; + } + + @Override + public GatewaySender create(String name, int remoteSystemId) { + this.name = name; + this.remoteSystemId = remoteSystemId; + when(gatewaySender.getId()).thenReturn(this.name); + when(gatewaySender.getRemoteDSId()).thenReturn(this.remoteSystemId); + when(gatewaySender.getAlertThreshold()).thenReturn(this.alertThreshold); + when(gatewaySender.getBatchSize()).thenReturn(this.batchSize); + when(gatewaySender.getBatchTimeInterval()).thenReturn(this.batchTimeInterval); + when(gatewaySender.getDiskStoreName()).thenReturn(this.diskStoreName); + when(gatewaySender.getDispatcherThreads()).thenReturn(this.dispatcherThreads); + when(gatewaySender.getGatewayEventFilters()).thenReturn(this.eventFilters); + when(gatewaySender.getGatewayTransportFilters()).thenReturn(this.transportFilters); + when(gatewaySender.getMaximumQueueMemory()).thenReturn(this.maxQueueMemory); + when(gatewaySender.getOrderPolicy()).thenReturn(this.orderPolicy); + when(gatewaySender.getSocketBufferSize()).thenReturn(this.socketBufferSize); + when(gatewaySender.getSocketReadTimeout()).thenReturn(this.socketReadTimeout); + when(gatewaySender.isManualStart()).thenReturn(this.manualStart); + when(gatewaySender.isBatchConflationEnabled()).thenReturn(this.batchConflationEnabled); + when(gatewaySender.isDiskSynchronous()).thenReturn(this.diskSynchronous); + when(gatewaySender.isParallel()).thenReturn(this.parallel); + when(gatewaySender.isPersistenceEnabled()).thenReturn(this.persistenceEnabled); + return gatewaySender; + } + + @Override + public GatewaySenderFactory removeGatewayEventFilter(GatewayEventFilter filter) { + gatewaySender.removeGatewayEventFilter(filter); + return this; + } + + @Override + public GatewaySenderFactory setAlertThreshold(int alertThreshold) { + this.alertThreshold = alertThreshold; + return this; + } + + @Override + public GatewaySenderFactory setBatchConflationEnabled(boolean batchConflationEnabled) { + this.batchConflationEnabled = batchConflationEnabled; + return this; + } + + @Override + public GatewaySenderFactory setBatchSize(int batchSize) { + this.batchSize = batchSize; + return this; + } + + @Override + public GatewaySenderFactory setBatchTimeInterval(int batchTimeInterval) { + this.batchTimeInterval = batchTimeInterval; + return this; + } + + @Override + public GatewaySenderFactory setDiskStoreName(String diskStoreName) { + this.diskStoreName = diskStoreName; + return this; + } + + @Override + public GatewaySenderFactory setDiskSynchronous(boolean diskSynchronous) { + this.diskSynchronous = diskSynchronous; + return this; + } + + @Override + public GatewaySenderFactory setDispatcherThreads(int dispatcherThreads) { + this.dispatcherThreads = dispatcherThreads; + return this; + } + + @Override + public GatewaySenderFactory setManualStart(boolean manualStart) { + this.manualStart = manualStart; + return this; + } + + @Override + public GatewaySenderFactory setMaximumQueueMemory(int maxQueueMemory) { + this.maxQueueMemory = maxQueueMemory; + return this; + } + + @Override + public GatewaySenderFactory setOrderPolicy(OrderPolicy orderPolicy) { + this.orderPolicy = orderPolicy; + return this; + } + + @Override + public GatewaySenderFactory setParallel(boolean parallel) { + this.parallel = parallel; + return this; + } + + @Override + public GatewaySenderFactory setPersistenceEnabled(boolean persistenceEnabled) { + this.persistenceEnabled = persistenceEnabled; + return this; + } + + @Override + public GatewaySenderFactory setSocketBufferSize(int socketBufferSize) { + this.socketBufferSize = socketBufferSize; + return this; + } + + @Override + public GatewaySenderFactory setSocketReadTimeout(int socketReadTimeout) { + this.socketReadTimeout = socketReadTimeout; + return this; + } + + @Override + public GatewaySenderFactory removeGatewayTransportFilter( + GatewayTransportFilter arg0) { + // TODO Auto-generated method stub + return null; + } + } + + + /* + * This mocks out the WAN components which are disabled in the developer edition + */ + @Override + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + if (bean instanceof GatewaySenderFactoryBean) { + ((GatewaySenderFactoryBean)bean).setFactory(new StubGWSenderFactory()); + } + if (bean instanceof AsyncEventQueueFactoryBean) { + ((AsyncEventQueueFactoryBean)bean).setFactory(new StubAsyncEventQueueFactory()); + } + return bean; + } + + @Override + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + return bean; + } +} diff --git a/src/test/resources/org/springframework/data/gemfire/config/cache-ns.xml b/src/test/resources/org/springframework/data/gemfire/config/cache-ns.xml index 2083b55f..fc410b67 100644 --- a/src/test/resources/org/springframework/data/gemfire/config/cache-ns.xml +++ b/src/test/resources/org/springframework/data/gemfire/config/cache-ns.xml @@ -29,5 +29,11 @@ + + + + + + - \ No newline at end of file + 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 new file mode 100644 index 00000000..fbdddb1a --- /dev/null +++ b/src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml @@ -0,0 +1,75 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file