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

upgraded to 1.2.1 - GemFire 7.0 support

Browse Source
This commit is contained in:
David Turanski
2012-10-09 16:18:27 -04:00
parent bff9f3ce7c 154905b808
commit e3de5510d3
16 changed files with 1653 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
build.gradle
View File

@@ -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"}
}
@@ -48,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"

4
gradle.properties
View File

@@ -7,7 +7,7 @@ slf4jVersion = 1.6.4
# Common libraries
springVersion = 3.1.2.RELEASE
springDataCommonsVersion = 1.4.0.RELEASE
gemfireVersion = 6.6.3
gemfireVersion = 7.0.Beta-SNAPSHOT
gemfireVersion6x = 6.6.3
# Testing
@@ -25,4 +25,4 @@ gemfire.range = "[6.5, 8.0)"
# --------------------
# Project wide version
# --------------------
version=1.2.0.BUILD-SNAPSHOT
version=1.2.1.BUILD-SNAPSHOT

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

@@ -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<JndiDataSource> 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<JndiDataSource> jndiDataSources) {
this.jndiDataSources = jndiDataSources;
}
}
}

53
src/main/java/org/springframework/data/gemfire/RegionFactoryBean.java
View File

@@ -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<K, V> extends RegionLookupFactoryBean<K, V> imple
private CacheWriter<K, V> cacheWriter;
private Object gatewaySenders[];
private Object asyncEventQueues[];
private RegionAttributes<K, V> attributes;
private Scope scope;
@@ -98,10 +104,9 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
Cache c = (Cache) cache;
if (attributes != null) {
if (attributes != null)
AttributesFactory.validateAttributes(attributes);
}
final RegionFactory<K, V> regionFactory = (attributes != null ? c.createRegionFactory(attributes) : c
.<K, V> createRegionFactory());
@@ -111,17 +116,33 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> 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<K, V> 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<K, V> extends RegionLookupFactoryBean<K, V> 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<K, V> extends RegionLookupFactoryBean<K, V> imple
protected boolean isNotPersistent() {
return persistent != null && !persistent;
}
}
}

54
src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java Normal file
View File

@@ -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");
}
}

48
src/main/java/org/springframework/data/gemfire/config/GatewayReceiverParser.java Normal file
View File

@@ -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);
}
}

67
src/main/java/org/springframework/data/gemfire/config/GatewaySenderParser.java Normal file
View File

@@ -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);
}
}

5
src/main/java/org/springframework/data/gemfire/config/GemfireNamespaceHandler.java
View File

@@ -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());
}
}
}

121
src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java Normal file
View File

@@ -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<AsyncEventQueue> {
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;
}
}

123
src/main/java/org/springframework/data/gemfire/wan/GatewayReceiverFactoryBean.java Normal file
View File

@@ -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<GatewayReceiver> {
private GatewayReceiver gatewayReceiver;
private List<GatewayTransportFilter> 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<GatewayTransportFilter> 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;
}
}

237
src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java Normal file
View File

@@ -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<GatewaySender> {
private static List<String> validOrderPolicyValues = Arrays.asList("KEY", "PARTITION", "THREAD");
private GatewaySender gatewaySender;
private int remoteDistributedSystemId;
private List<GatewayEventFilter> eventFilters;
private List<GatewayTransportFilter> 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<GatewayEventFilter> gatewayEventFilters) {
this.eventFilters = gatewayEventFilters;
}
public void setTransportFilters(List<GatewayTransportFilter> 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;
}
}

324
src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.2.xsd
View File

@@ -35,7 +35,45 @@ and may be nested or referenced.
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="gateway-conflict-resolver"
minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation
source="com.gemstone.gemfire.cache.util.GatewayConflictResolver"><![CDATA[
A gateway conflict resolver for this cache. A gateway conflict resolver handles conflicts in the case of concurrent updates using a WAN gateway. The bean
must implement com.gemstone.gemfire.cache.util.GatewayConflictResolver. Requires Gemfire version 7.0 or higher.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:exports
type="com.gemstone.gemfire.cache.util.GatewayConflictResolver" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:any namespace="##other"
processContents="skip" minOccurs="0"
maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation><![CDATA[
Inner bean definition of the gateway conflict resolver.
]]></xsd:documentation>
</xsd:annotation>
</xsd:any>
</xsd:sequence>
<xsd:attribute name="ref" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The name of the gateway conflict resolver bean referred by this declaration. Used as a convenience method. If no reference exists,
use inner bean declarations.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="dynamic-region-factory"
minOccurs="0" maxOccurs="1">
<xsd:annotation>
@@ -468,6 +506,40 @@ when one or more missing required roles is restored to the distributed membershi
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:choice minOccurs="0" maxOccurs="unbounded">
<xsd:element name="gateway-sender"
type="baseGatewaySenderType" />
<xsd:element name="gateway-sender-ref">
<xsd:complexType>
<xsd:attribute name="bean"
type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The name of the gateway sender bean referred by this declaration. Used as a convenience method. If no reference exists,
use inner bean declarations.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:choice>
<xsd:choice minOccurs="0" maxOccurs="unbounded">
<xsd:element name="async-event-queue"
type="baseAsyncEventQueueType" />
<xsd:element name="async-event-queue-ref">
<xsd:complexType>
<xsd:attribute name="bean"
type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The name of the gateway sender bean referred by this declaration. Used as a convenience method. If no reference exists,
use inner bean declarations.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:choice>
</xsd:sequence>
<xsd:attribute name="persistent" type="xsd:string">
<xsd:annotation>
@@ -591,7 +663,7 @@ The fully qualified class name of the expected value type
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies if WAN gateway communications are enabled for this region (true or false)
Specifies if WAN gateway communications are enabled for this region (true or false) (Deprecated since Gemfire v 7.0)
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -599,7 +671,7 @@ The fully qualified class name of the expected value type
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies if WAN gateway hub id if enable-gateway is true.
Specifies if WAN gateway hub id if enable-gateway is true. (Deprecated since Gemfire v 7.0)
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -2130,6 +2202,137 @@ Specifies a data type if other than java.lang.String
</xsd:attribute>
</xsd:complexType>
<!-- -->
<xsd:complexType name="baseGatewaySenderType">
<xsd:annotation>
<xsd:documentation
source="com.gemstone.gemfire.cache.wan.GatewaySender"><![CDATA[
A gateway sender gateway definition (requires Gemfire 7.0 or later)
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:exports
type="com.gemstone.gemfire.cache.wan.GatewaySender" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
<xsd:sequence>
<xsd:element name="event-filter" minOccurs="0"
maxOccurs="1">
<xsd:annotation>
<xsd:documentation
source="com.gemstone.gemfire.cache.wan.GatewayEventFilter"><![CDATA[
A gateway event filter for this gateway sender
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:exports
type="com.gemstone.gemfire.cache.wan.GatewayEventFilter" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:any namespace="##other"
processContents="skip" minOccurs="0"
maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation><![CDATA[
Inner bean definition of the event filter
]]></xsd:documentation>
</xsd:annotation>
</xsd:any>
</xsd:sequence>
</xsd:complexType>
</xsd:element>
<xsd:element name="transport-filter" type="gatewayTransportFilterType"
minOccurs="0" maxOccurs="1" />
</xsd:sequence>
<xsd:attributeGroup ref="commonWANQueueAttributes" />
<xsd:attribute name="remote-distributed-system-id"
type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the remote distributed system id, an integer value representing the remote distributed system
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="manual-start" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies if the gateway sender is manually (true) or automatically(false) started
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="socket-buffer-size" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the socket buffer size in bytes
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="socket-read-timeout" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the socket read timeout in milliseconds
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="enable-batch-conflation"
type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies whether batch conflation is enabled (true or false)
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="batch-time-interval" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The maximum time interval that can elapse before a partial batch is sent from a GatewaySender to its corresponding GatewayReceiver.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="alert-threshold" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the alert threshold in miliseconds, indicating the maximum time elapsed from when the gateway sent the message
to when the acknowldgement was received from the gateway receiver.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="dispatcher-threads" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the number of dispatcher threads to allocate to the gateway sender
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order-policy" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the order policy - This only applies if parallel is enabled:
KEY: Indicates that events will be parallelized based on the event's key,
PARTITION:Indicates that events will be parallelized based on the event's: partition (using the PartitionResolver)
THREAD:Indicates that events will be parallelized based on the event's originating member and thread
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="parallel" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies whether all VMs need to distribute events to remote site. In this case only the events originating in a particular VM will be dispatched in order.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<!-- -->
<xsd:attributeGroup name="commonWANQueueAttributes">
<xsd:attribute name="batch-size" type="xsd:string"
@@ -2240,6 +2443,113 @@ The id of the cache - default is gemfireCache
</xsd:attribute>
</xsd:complexType>
<!-- -->
<xsd:complexType name="baseAsyncEventQueueType">
<xsd:annotation>
<xsd:documentation
source="com.gemstone.gemfire.cache.wan.AsyncEventQueue"><![CDATA[
An async event queue definition (requires Gemfire 7.0 or later)
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:exports
type="com.gemstone.gemfire.cache.wan.AsyncEventQueue" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
<xsd:sequence>
<xsd:element name="async-event-listener"
minOccurs="1" maxOccurs="1">
<xsd:annotation>
<xsd:documentation
source="com.gemstone.gemfire.cache.wan.AsyncEventListener"><![CDATA[
An async event listener definition for this distributed system. (requires Gemfire 7.0)
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:exports
type="com.gemstone.gemfire.cache.wan.AsyncEventListener" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:any namespace="##other"
processContents="skip" minOccurs="0"
maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation><![CDATA[
Inner bean definition of the async event listener
]]></xsd:documentation>
</xsd:annotation>
</xsd:any>
</xsd:sequence>
<xsd:attribute name="ref" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The name of the async event listener bean referred by this declaration. Used as a convenience method. If no reference exists,
use inner bean declarations.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:sequence>
<xsd:attributeGroup ref="commonWANQueueAttributes" />
</xsd:complexType>
<!-- -->
<xsd:element name="gateway-sender">
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="baseGatewaySenderType">
<xsd:attribute name="id" type="xsd:string"
use="required">
<xsd:annotation>
<xsd:documentation><![CDATA[
The id of this bean definition.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="cache-ref" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The id of the cache - default is gemfireCache
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<!-- -->
<xsd:element name="async-event-queue">
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="baseAsyncEventQueueType">
<xsd:attribute name="id" type="xsd:string"
use="required">
<xsd:annotation>
<xsd:documentation><![CDATA[
The id of this bean definition.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="cache-ref" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
The id of the cache - default is gemfireCache
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="gateway-receiver" type="gatewayReceiverType" />
<!-- -->
<xsd:element name="function-service">
<xsd:complexType>
<xsd:sequence>
@@ -2342,6 +2652,7 @@ use inner bean declarations.
<xsd:complexType name="gatewayHubType">
<xsd:annotation>
<xsd:documentation><![CDATA[
Deprecated as of Gemfire 7
]]></xsd:documentation>
</xsd:annotation>
<xsd:sequence>
@@ -2349,6 +2660,7 @@ use inner bean declarations.
minOccurs="0" maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation><![CDATA[
Deprecated as of Gemfire 7
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
@@ -2412,6 +2724,7 @@ Specifies the startup policy (primary,secondary, none) for the gateway hub
<xsd:complexType name="gatewayEndpointType">
<xsd:annotation>
<xsd:documentation><![CDATA[
Deprecated as of Gemfire 7
]]></xsd:documentation>
</xsd:annotation>
<xsd:attribute name="host" type="xsd:string" use="required">
@@ -2426,6 +2739,7 @@ Specifies the startup policy (primary,secondary, none) for the gateway hub
<xsd:complexType name="gatewayQueueType">
<xsd:annotation>
<xsd:documentation><![CDATA[
Deprecated as of Gemfire 7
]]></xsd:documentation>
</xsd:annotation>
<xsd:attribute name="enable-batch-conflation"
@@ -2490,6 +2804,7 @@ Specifies the maximum memory in MB to allocate for the queue
<xsd:complexType name="gatewayType">
<xsd:annotation>
<xsd:documentation><![CDATA[
Deprecated as of Gemfire 7
]]></xsd:documentation>
</xsd:annotation>
<xsd:sequence>
@@ -2585,8 +2900,9 @@ Specifies the number of parallel threads
<xsd:element name="gateway-hub" type="gatewayHubType">
<xsd:annotation>
<xsd:documentation><![CDATA[
Deprecated as of Gemfire 7
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
<!-- End Gemfire 6 WAN Gateway schema -->
</xsd:schema>
</xsd:schema>

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

@@ -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);
}
}
public static class TestConflictResolver implements GatewayConflictResolver {
@Override
public void onEvent(TimestampedEntryEvent arg0, GatewayConflictHelper arg1) {
// TODO Auto-generated method stub
}
}
}

507
src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java Normal file
View File

@@ -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<GatewayEventFilter> eventFilters = TestUtils.readField("eventFilters", gwsfb);
assertNotNull(eventFilters);
assertEquals(2, eventFilters.size());
assertTrue(eventFilters.get(0) instanceof TestEventFilter);
List<GatewayTransportFilter> 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<String> gatewaySenders = region.getAttributes().getGatewaySenderIds();
// assertTrue(gatewaySenders.remove(gws));
// gatewaySenders.remove(gws);
RegionFactoryBean rfb = ctx.getBean("&region-inner-gateway-sender", RegionFactoryBean.class);
Object[] gwsenders = TestUtils.readField("gatewaySenders", rfb);
gws = (GatewaySender)gwsenders[0];
List<GatewayEventFilter> eventFilters = gws.getGatewayEventFilters();
assertNotNull(eventFilters);
assertEquals(1, eventFilters.size());
assertTrue(eventFilters.get(0) instanceof TestEventFilter);
List<GatewayTransportFilter> 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<AsyncEvent> 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<GatewayEventFilter> eventFilters;
private List<GatewayTransportFilter> transportFilters;
private String name;
private int remoteSystemId;
public StubGWSenderFactory() {
this.eventFilters = new ArrayList<GatewayEventFilter>();
this.transportFilters = new ArrayList<GatewayTransportFilter>();
}
@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;
}
}

8
src/test/resources/org/springframework/data/gemfire/config/cache-ns.xml
View File

@@ -33,5 +33,11 @@
</gfe:pool>
<gfe:cache id="heap-tuned-cache" critical-heap-percentage="70.0" eviction-heap-percentage="60.0"/>
<gfe:cache id="cache-with-conflict-resolver">
<gfe:gateway-conflict-resolver>
<bean class="org.springframework.data.gemfire.config.CacheNamespaceTest.TestConflictResolver"/>
</gfe:gateway-conflict-resolver>
</gfe:cache>
</beans>

75
src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml Normal file
View File

@@ -0,0 +1,75 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:gfe="http://www.springframework.org/schema/gemfire"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:util="http://www.springframework.org/schema/util"
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"
disk-store-ref="diskstore"
enable-batch-conflation="true"
maximum-queue-memory="50"
order-policy="THREAD"
persistent="true"
parallel="true"
socket-buffer-size="16536"
socket-read-timeout="3000">
<gfe:event-filter>
<bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestEventFilter"/>
</gfe:event-filter>
<gfe:transport-filter>
<bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestTransportFilter"/>
</gfe:transport-filter>
</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-listener>
<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"
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"
remote-distributed-system-id="2"
alert-threshold="10"
batch-size="11"
dispatcher-threads="12"
manual-start="true">
<gfe:event-filter>
<ref bean="event-filter"/>
<bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestEventFilter"/>
</gfe:event-filter>
<gfe:transport-filter>
<ref bean="transport-filter"/>
<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>