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

Added WAN support modifications for GF 7 and general cleanup of test code

Browse Source
This commit is contained in:
David Turanski
2012-07-23 14:41:00 -04:00
parent 0d84ef62ad
commit 94468bec9a
60 changed files with 4707 additions and 1670 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,7 +17,6 @@
package org.springframework.data.gemfire;
import java.lang.reflect.Field;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -28,6 +27,7 @@ import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import org.springframework.util.ReflectionUtils;
import com.gemstone.gemfire.cache.AsyncEventQueue;
import com.gemstone.gemfire.cache.AttributesFactory;
import com.gemstone.gemfire.cache.Cache;
import com.gemstone.gemfire.cache.CacheClosedException;
@@ -39,11 +39,12 @@ 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;
/**
* FactoryBean for creating generic GemFire {@link Region}s. Will try to first
* locate the region (by name) and, in case none if found, proceed to creating
* one using the given settings.
* Base class for FactoryBeans used to create GemFire {@link Region}s. Will try
* to first locate the region (by name) and, in case none if found, proceed to
* creating one using the given settings.
*
* Note that this factory bean allows for very flexible creation of GemFire
* {@link Region}. For "client" regions however, see
@@ -69,20 +70,26 @@ public abstract class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K,
private CacheWriter<K, V> cacheWriter;
private GatewaySender gatewaySenders[];
private AsyncEventQueue asyncEventQueues[];
private RegionAttributes<K, V> attributes;
private Scope scope;
private boolean persistent;
private Boolean enableGateway;
private String hubId;
private String diskStoreName;
private String dataPolicyName;
private Region<K, V> region;
private List<Region<?, ?>> subRegions;
@Override
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
@@ -102,12 +109,39 @@ public abstract class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K,
final RegionFactory<K, V> regionFactory = (attributes != null ? c.createRegionFactory(attributes) : c
.<K, V> createRegionFactory());
if (hubId != null) {
enableGateway = enableGateway == null ? true : enableGateway;
Assert.isTrue(enableGateway, "hubId requires the enableGateway property to be true");
regionFactory.setGatewayHubId(hubId);
}
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 (GatewaySender gatewaySender : gatewaySenders) {
regionFactory.addGatewaySender(gatewaySender);
}
}
if (!ObjectUtils.isEmpty(asyncEventQueues)) {
for (AsyncEventQueue asyncEventQueue : asyncEventQueues) {
regionFactory.addAsyncEventQueue(asyncEventQueue);
}
}
if (cacheLoader != null) {
regionFactory.setCacheLoader(cacheLoader);
}
@@ -256,11 +290,6 @@ public abstract class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K,
this.cacheListeners = cacheListeners;
}
public void setSubRegions(List<Region<?, ?>> subRegions) {
log.info("setting subRegions");
this.subRegions = subRegions;
}
/**
* Sets the cache loader used for the region used by this factory. Used only
* when a new region is created.Overrides the settings specified through
@@ -315,6 +344,22 @@ public abstract class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K,
this.diskStoreName = diskStoreName;
}
public void setGatewaySenders(GatewaySender[] gatewaySenders) {
this.gatewaySenders = gatewaySenders;
}
public void setAsyncEventQueues(AsyncEventQueue[] asyncEventQueues) {
this.asyncEventQueues = asyncEventQueues;
}
public void setEnableGateway(boolean enableGateway) {
this.enableGateway = enableGateway;
}
public void setHubId(String hubId) {
this.hubId = hubId;
}
/**
* Sets the region attributes used for the region used by this factory.
* Allows maximum control in specifying the region settings. Used only when

81
src/main/java/org/springframework/data/gemfire/config/AbstractRegionParser.java
View File

@@ -23,7 +23,9 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.ManagedArray;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
@@ -117,28 +119,67 @@ abstract class AbstractRegionParser extends AliasReplacingBeanDefinitionParser {
ParsingUtils.parseEviction(parserContext, element, attrBuilder);
ParsingUtils.parseMembershipAttributes(parserContext, element, attrBuilder);
String enableGateway = element.getAttribute("enable-gateway");
String hubId = element.getAttribute("hub-id");
// Factory will enable gateway if it is not set and hub-id is set.
if (StringUtils.hasText(enableGateway)) {
if (ParsingUtils.GEMFIRE_VERSION.startsWith("7")) {
log.warn("'enable-gateway' is deprecated since Gemfire 7.0");
}
}
ParsingUtils.setPropertyValue(element, builder, "enable-gateway");
if (StringUtils.hasText(hubId)) {
if (ParsingUtils.GEMFIRE_VERSION.startsWith("7")) {
log.warn("'hub-id' is deprecated since Gemfire 7.0");
}
if (!CollectionUtils.isEmpty(DomUtils.getChildElementsByTagName(element, "gateway-sender"))) {
parserContext.getReaderContext().error("It is invalid to specify both 'hub-id' and 'gateway-sender'",
element);
}
}
ParsingUtils.setPropertyValue(element, builder, "hub-id");
// Parse child elements
parseCollectionOfCustomSubElements(parserContext, element, builder,
"com.gemstone.gemfire.cache.wan.GatewaySender", "gateway-sender", "gatewaySenders");
parseCollectionOfCustomSubElements(parserContext, element, builder,
"com.gemstone.gemfire.cache.wan.AsyncEventQueue", "async-event-queue", "asyncEventQueues");
List<Element> subElements = DomUtils.getChildElements(element);
// parse nested elements
for (Element subElement : subElements) {
String name = subElement.getLocalName();
if ("cache-listener".equals(name)) {
builder.addPropertyValue("cacheListeners", parseCacheListener(parserContext, subElement, builder));
if (subElement.getLocalName().equals("cache-listener")) {
builder.addPropertyValue("cacheListeners",
ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, subElement, builder));
}
else if ("cache-loader".equals(name)) {
builder.addPropertyValue("cacheLoader", parseCacheLoader(parserContext, subElement, builder));
else if (subElement.getLocalName().equals("cache-loader")) {
builder.addPropertyValue("cacheLoader",
ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, subElement, builder));
}
else if ("cache-writer".equals(name)) {
builder.addPropertyValue("cacheWriter", parseCacheWriter(parserContext, subElement, builder));
else if (subElement.getLocalName().equals("cache-writer")) {
builder.addPropertyValue("cacheWriter",
ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, subElement, builder));
}
// subregion
else if (name.endsWith("region")) {
else if (subElement.getLocalName().endsWith("region")) {
doParseSubRegion(element, subElement, parserContext, builder, subRegion);
}
}
}
private void parseCollectionOfCustomSubElements(ParserContext parserContext, Element element,
BeanDefinitionBuilder builder, String className, String subElementName, String propertyName) {
List<Element> subElements = DomUtils.getChildElementsByTagName(element, new String[] { subElementName,
subElementName + "-ref" });
if (!CollectionUtils.isEmpty(subElements)) {
ManagedArray array = new ManagedArray(className, subElements.size());
for (Element subElement : subElements) {
array.add(ParsingUtils.parseRefOrNestedCustomElement(parserContext, subElement, builder));
}
builder.addPropertyValue(propertyName, array);
}
}
private BeanDefinition parseSubRegion(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
@@ -151,16 +192,4 @@ abstract class AbstractRegionParser extends AliasReplacingBeanDefinitionParser {
String name = element.getAttribute(NAME_ATTRIBUTE);
return StringUtils.hasText(name) ? name : element.getAttribute(ID_ATTRIBUTE);
}
private Object parseCacheListener(ParserContext parserContext, Element subElement, BeanDefinitionBuilder builder) {
return ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, subElement, builder);
}
private Object parseCacheLoader(ParserContext parserContext, Element subElement, BeanDefinitionBuilder builder) {
return ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, subElement, builder);
}
private Object parseCacheWriter(ParserContext parserContext, Element subElement, BeanDefinitionBuilder builder) {
return ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, subElement, builder);
}
}

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")) ? "gemfire-cache" : 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");
}
}

101
src/main/java/org/springframework/data/gemfire/config/GatewayHubParser.java Normal file
View File

@@ -0,0 +1,101 @@
/*
* 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 java.util.List;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.data.gemfire.wan.GatewayHubFactoryBean;
import org.springframework.data.gemfire.wan.GatewayProxy;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
/**
* @author David Turanski
*
*/
class GatewayHubParser extends AbstractSimpleBeanDefinitionParser {
@Override
protected Class<?> getBeanClass(Element element) {
return GatewayHubFactoryBean.class;
}
@SuppressWarnings({ "rawtypes", "unchecked" })
@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 : "gemfire-cache"));
ParsingUtils.setPropertyValue(element, builder, "bind-address");
ParsingUtils.setPropertyValue(element, builder, "manual-start");
ParsingUtils.setPropertyValue(element, builder, "socket-buffer-size");
ParsingUtils.setPropertyValue(element, builder, "startup-policy");
List<Element> gatewayElements = DomUtils.getChildElementsByTagName(element, "gateway");
if (!CollectionUtils.isEmpty(gatewayElements)) {
ManagedList gateways = new ManagedList();
for (Element gatewayElement : gatewayElements) {
BeanDefinitionBuilder gatewayBuilder = BeanDefinitionBuilder.genericBeanDefinition(GatewayProxy.class);
ParsingUtils.setPropertyValue(gatewayElement, gatewayBuilder, "gateway-id", "id");
ParsingUtils.setPropertyValue(gatewayElement, gatewayBuilder, "concurrency-level");
ParsingUtils.setPropertyValue(gatewayElement, gatewayBuilder, "socket-read-timeout");
ParsingUtils.setPropertyValue(gatewayElement, gatewayBuilder, "socket-buffer-size");
ParsingUtils.setPropertyValue(gatewayElement, gatewayBuilder, "order-policy");
List<Element> endpointElements = DomUtils.getChildElementsByTagName(gatewayElement, "gateway-endpoint");
if (!CollectionUtils.isEmpty(endpointElements)) {
ManagedList endpoints = new ManagedList();
for (Element endpointElement : endpointElements) {
BeanDefinitionBuilder endpointBuilder = BeanDefinitionBuilder
.genericBeanDefinition(GatewayProxy.GatewayEndpoint.class);
ParsingUtils.setPropertyValue(endpointElement, endpointBuilder, "host");
ParsingUtils.setPropertyValue(endpointElement, endpointBuilder, "port");
ParsingUtils.setPropertyValue(endpointElement, endpointBuilder, "endpoint-id", "id");
endpoints.add(endpointBuilder.getBeanDefinition());
}
gatewayBuilder.addPropertyValue("endpoints", endpoints);
}
Element gatewayListenerElement = DomUtils.getChildElementByTagName(gatewayElement, "gateway-listener");
if (gatewayListenerElement != null) {
Object obj = ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, gatewayListenerElement,
gatewayBuilder);
gatewayBuilder.addPropertyValue("listeners", obj);
}
Element gatewayQueueElement = DomUtils.getChildElementByTagName(gatewayElement, "gateway-queue");
if (gatewayQueueElement != null) {
BeanDefinitionBuilder queueBuilder = BeanDefinitionBuilder
.genericBeanDefinition(GatewayProxy.GatewayQueue.class);
ParsingUtils.setPropertyValue(gatewayQueueElement, queueBuilder, "alert-threshold");
ParsingUtils.setPropertyValue(gatewayQueueElement, queueBuilder, "batch-size");
ParsingUtils.setPropertyValue(gatewayQueueElement, queueBuilder, "batch-time-interval");
ParsingUtils.setPropertyValue(gatewayQueueElement, queueBuilder, "maximum-queue-memory");
ParsingUtils.setPropertyValue(gatewayQueueElement, queueBuilder, "persistent");
ParsingUtils.setPropertyValue(gatewayQueueElement, queueBuilder, "disk-store-ref");
ParsingUtils.setPropertyValue(gatewayQueueElement, queueBuilder, "enable-batch-conflation");
gatewayBuilder.addPropertyValue("queue", queueBuilder.getBeanDefinition());
}
gateways.add(gatewayBuilder.getBeanDefinition());
}
builder.addPropertyValue("gateways", gateways);
}
}
}

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 : "gemfire-cache"));
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 : "gemfire-cache"));
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);
}
}

34
src/main/java/org/springframework/data/gemfire/config/GemfireDataNamespaceHandler.java Normal file
View File

@@ -0,0 +1,34 @@
/*
* 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.xml.NamespaceHandlerSupport;
import org.springframework.data.gemfire.repository.config.GemfireRepositoryParser;
/**
* Namespace handler for GemFire definitions.
*
* @author Costin Leau
* @author David Turanski
*/
class GemfireDataNamespaceHandler extends NamespaceHandlerSupport {
@Override
public void init() {
registerBeanDefinitionParser("repositories", new GemfireRepositoryParser());
}
}

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

@@ -16,8 +16,13 @@
package org.springframework.data.gemfire.config;
import java.util.Arrays;
import java.util.List;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
import org.springframework.data.gemfire.repository.config.GemfireRepositoryParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.w3c.dom.Element;
/**
* Namespace handler for GemFire definitions.
@@ -26,12 +31,25 @@ import org.springframework.data.gemfire.repository.config.GemfireRepositoryParse
* @author David Turanski
*/
class GemfireNamespaceHandler extends NamespaceHandlerSupport {
static final List<String> GEMFIRE7_ELEMENTS = Arrays.asList("async-event-queue", "gateway-sender",
"gateway-receiver");
@Override
public BeanDefinition parse(Element element, ParserContext parserContext) {
boolean v7ElementsPresent = GEMFIRE7_ELEMENTS.contains(element.getLocalName());
if (v7ElementsPresent) {
ParsingUtils.throwExceptionIfNotGemfireV7(element.getLocalName(), null, parserContext);
}
return super.parse(element, parserContext);
}
@Override
public void init() {
registerBeanDefinitionParser("cache", new CacheParser());
registerBeanDefinitionParser("client-cache", new ClientCacheParser());
registerBeanDefinitionParser("lookup-region", new LookupRegionParser());
registerBeanDefinitionParser("replicated-region", new ReplicatedRegionParser());
registerBeanDefinitionParser("partitioned-region", new PartitionedRegionParser());
@@ -43,6 +61,10 @@ class GemfireNamespaceHandler extends NamespaceHandlerSupport {
registerBeanDefinitionParser("cache-server", new CacheServerParser());
registerBeanDefinitionParser("transaction-manager", new TransactionManagerParser());
registerBeanDefinitionParser("cq-listener-container", new GemfireListenerContainerParser());
registerBeanDefinitionParser("repositories", new GemfireRepositoryParser());
registerBeanDefinitionParser("async-event-queue", new AsyncEventQueueParser());
registerBeanDefinitionParser("gateway-sender", new GatewaySenderParser());
registerBeanDefinitionParser("gateway-receiver", new GatewayReceiverParser());
// V6 WAN parsers
registerBeanDefinitionParser("gateway-hub", new GatewayHubParser());
}
}

77
src/main/java/org/springframework/data/gemfire/config/ParsingUtils.java
View File

@@ -33,6 +33,7 @@ import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
import com.gemstone.gemfire.cache.CacheFactory;
import com.gemstone.gemfire.cache.ExpirationAction;
import com.gemstone.gemfire.cache.ExpirationAttributes;
import com.gemstone.gemfire.cache.LossAction;
@@ -48,6 +49,8 @@ import com.gemstone.gemfire.cache.Scope;
*/
abstract class ParsingUtils {
final static String GEMFIRE_VERSION = CacheFactory.getVersion();
private static final String ALIASES_KEY = ParsingUtils.class.getName() + ":aliases";
static void setPropertyValue(Element element, BeanDefinitionBuilder builder, String attributeName,
@@ -119,11 +122,14 @@ abstract class ParsingUtils {
*/
static Object parseRefOrNestedBeanDeclaration(ParserContext parserContext, Element element,
BeanDefinitionBuilder builder) {
return parseRefOrNestedBeanDeclaration(parserContext, element, builder, "ref");
return parseRefOrNestedBeanDeclaration(parserContext, element, builder, "ref", false);
}
static Object parseRefOrNestedBeanDeclaration(ParserContext parserContext, Element element,
BeanDefinitionBuilder builder, String refAttrName) {
static Object getBeanReference(ParserContext parserContext, Element element) {
return getBeanReference(parserContext, element, "ref");
}
static Object getBeanReference(ParserContext parserContext, Element element, String refAttrName) {
String attr = element.getAttribute(refAttrName);
boolean hasRef = StringUtils.hasText(attr);
@@ -138,18 +144,54 @@ abstract class ParsingUtils {
}
return new RuntimeBeanReference(attr);
}
if (childElements.isEmpty()) {
parserContext.getReaderContext().error(
"specify either '" + refAttrName + "' attribute or a nested bean declaration for '"
+ element.getLocalName() + "' element", element);
else {
return null;
}
}
static Object parseRefOrNestedCustomElement(ParserContext parserContext, Element element,
BeanDefinitionBuilder builder) {
Object beanRef = ParsingUtils.getBeanReference(parserContext, element, "bean");
if (beanRef != null) {
return beanRef;
}
else {
return parserContext.getDelegate().parseCustomElement(element, builder.getBeanDefinition());
}
}
static Object parseRefOrSingleNestedBeanDeclaration(ParserContext parserContext, Element element,
BeanDefinitionBuilder builder) {
return parseRefOrNestedBeanDeclaration(parserContext, element, builder, "ref", true);
}
static Object parseRefOrNestedBeanDeclaration(ParserContext parserContext, Element element,
BeanDefinitionBuilder builder, String refAttrName) {
return parseRefOrNestedBeanDeclaration(parserContext, element, builder, refAttrName, false);
}
static Object parseRefOrNestedBeanDeclaration(ParserContext parserContext, Element element,
BeanDefinitionBuilder builder, String refAttrName, boolean single) {
Object beanRef = getBeanReference(parserContext, element, refAttrName);
if (beanRef != null) {
return beanRef;
}
// check nested declarations
List<Element> childElements = DomUtils.getChildElements(element);
// nested parse nested bean definition
if (childElements.size() == 1) {
return parserContext.getDelegate().parsePropertySubElement(childElements.get(0),
builder.getRawBeanDefinition());
}
else {
if (single) {
parserContext.getReaderContext().error(
"the element '" + element.getLocalName()
+ "' does not support multiple nested bean definitions", element);
}
}
ManagedList<Object> list = new ManagedList<Object>();
@@ -199,6 +241,14 @@ abstract class ParsingUtils {
return true;
}
static void parseTransportFilters(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
Element transportFilterElement = DomUtils.getChildElementByTagName(element, "transport-filter");
if (transportFilterElement != null) {
builder.addPropertyValue("transportFilters",
parseRefOrNestedBeanDeclaration(parserContext, transportFilterElement, builder));
}
}
static void parseStatistics(Element element, BeanDefinitionBuilder attrBuilder) {
setPropertyValue(element, attrBuilder, "statistics", "statisticsEnabled");
}
@@ -276,6 +326,17 @@ abstract class ParsingUtils {
}
}
static void throwExceptionIfNotGemfireV7(String elementName, String attributeName, ParserContext parserContext) {
boolean checkGemfireV7 = !GEMFIRE_VERSION.startsWith("7");
if (checkGemfireV7) {
String messagePrefix = (attributeName == null) ? "element '" + elementName + "'" : "attribute '"
+ attributeName + " of element '" + elementName + "'";
parserContext.getReaderContext().error(
messagePrefix + " requires Gemfire version 7 or later. The current version is " + GEMFIRE_VERSION,
null);
}
}
static void parseScope(Element element, BeanDefinitionBuilder builder) {
String scope = element.getAttribute("scope");
if (StringUtils.hasText(scope)) {

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

@@ -42,11 +42,24 @@ public class GemfirePersistentProperty extends AnnotationBasedPersistentProperty
super(field, propertyDescriptor, owner, simpleTypeHolder);
}
/* (non-Javadoc)
* @see org.springframework.data.mapping.model.AbstractPersistentProperty#createAssociation()
/*
* (non-Javadoc)
*
* @see org.springframework.data.mapping.model.AbstractPersistentProperty#
* createAssociation()
*/
@Override
protected Association<GemfirePersistentProperty> createAssociation() {
return null;
return new Association<GemfirePersistentProperty>(this, null);
}
@Override
public boolean isAssociation() {
return field.isAnnotationPresent(RegionRef.class) || super.isAssociation();
}
public RegionRef geRegionRef() {
return getField().getAnnotation(RegionRef.class);
}
}

49
src/main/java/org/springframework/data/gemfire/mapping/MappingPdxSerializer.java
View File

@@ -21,6 +21,7 @@ import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.data.convert.EntityInstantiator;
import org.springframework.data.convert.EntityInstantiators;
import org.springframework.data.mapping.PersistentEntity;
@@ -37,21 +38,24 @@ import com.gemstone.gemfire.pdx.PdxSerializer;
import com.gemstone.gemfire.pdx.PdxWriter;
/**
* {@link PdxSerializer} implementation that uses a {@link GemfireMappingContext} to read and write entities.
* {@link PdxSerializer} implementation that uses a
* {@link GemfireMappingContext} to read and write entities.
*
* @author Oliver Gierke
*/
public class MappingPdxSerializer implements PdxSerializer, ApplicationContextAware {
private final GemfireMappingContext mappingContext;
private final ConversionService conversionService;
private EntityInstantiators instantiators;
private SpELContext context;
/**
* Creates a new {@link MappingPdxSerializer} using the given {@link GemfireMappingContext} and
* {@link ConversionService}.
* Creates a new {@link MappingPdxSerializer} using the given
* {@link GemfireMappingContext} and {@link ConversionService}.
*
* @param mappingContext must not be {@literal null}.
* @param conversionService must not be {@literal null}.
@@ -68,7 +72,16 @@ public class MappingPdxSerializer implements PdxSerializer, ApplicationContextAw
}
/**
* Configures the {@link EntityInstantiator}s to be used to create the instances to be read.
* Creates a new {@link MappingPdxSerializer} using the default
* {@link GemfireMappingContext} and {@link DefaultConversionService}.
*/
public MappingPdxSerializer() {
this(new GemfireMappingContext(), new DefaultConversionService());
}
/**
* Configures the {@link EntityInstantiator}s to be used to create the
* instances to be read.
*
* @param gemfireInstantiators must not be {@literal null}.
*/
@@ -77,9 +90,12 @@ public class MappingPdxSerializer implements PdxSerializer, ApplicationContextAw
this.instantiators = new EntityInstantiators(gemfireInstantiators);
}
/*
/*
* (non-Javadoc)
* @see org.springframework.context.ApplicationContextAware#setApplicationContext(org.springframework.context.ApplicationContext)
*
* @see
* org.springframework.context.ApplicationContextAware#setApplicationContext
* (org.springframework.context.ApplicationContext)
*/
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
@@ -88,8 +104,11 @@ public class MappingPdxSerializer implements PdxSerializer, ApplicationContextAw
/*
* (non-Javadoc)
* @see com.gemstone.gemfire.pdx.PdxSerializer#fromData(java.lang.Class, com.gemstone.gemfire.pdx.PdxReader)
*
* @see com.gemstone.gemfire.pdx.PdxSerializer#fromData(java.lang.Class,
* com.gemstone.gemfire.pdx.PdxReader)
*/
@Override
public Object fromData(Class<?> type, final PdxReader reader) {
final GemfirePersistentEntity<?> entity = mappingContext.getPersistentEntity(type);
@@ -102,9 +121,11 @@ public class MappingPdxSerializer implements PdxSerializer, ApplicationContextAw
Object instance = instantiator.createInstance(entity, provider);
final BeanWrapper<PersistentEntity<Object, ?>, Object> wrapper = BeanWrapper.create(instance, conversionService);
final BeanWrapper<PersistentEntity<Object, ?>, Object> wrapper = BeanWrapper
.create(instance, conversionService);
entity.doWithProperties(new PropertyHandler<GemfirePersistentProperty>() {
@Override
public void doWithPersistentProperty(GemfirePersistentProperty persistentProperty) {
if (entity.isConstructorArgument(persistentProperty)) {
@@ -115,7 +136,8 @@ public class MappingPdxSerializer implements PdxSerializer, ApplicationContextAw
try {
wrapper.setProperty(persistentProperty, value);
} catch (Exception e) {
}
catch (Exception e) {
throw new MappingException("Could not read value " + value.toString(), e);
}
}
@@ -126,20 +148,25 @@ public class MappingPdxSerializer implements PdxSerializer, ApplicationContextAw
/*
* (non-Javadoc)
* @see com.gemstone.gemfire.pdx.PdxSerializer#toData(java.lang.Object, com.gemstone.gemfire.pdx.PdxWriter)
*
* @see com.gemstone.gemfire.pdx.PdxSerializer#toData(java.lang.Object,
* com.gemstone.gemfire.pdx.PdxWriter)
*/
@Override
public boolean toData(Object value, final PdxWriter writer) {
GemfirePersistentEntity<?> entity = mappingContext.getPersistentEntity(value.getClass());
final BeanWrapper<PersistentEntity<Object, ?>, Object> wrapper = BeanWrapper.create(value, conversionService);
entity.doWithProperties(new PropertyHandler<GemfirePersistentProperty>() {
@Override
public void doWithPersistentProperty(GemfirePersistentProperty persistentProperty) {
try {
Object value = wrapper.getProperty(persistentProperty);
writer.writeObject(persistentProperty.getName(), value);
} catch (Exception e) {
}
catch (Exception e) {
throw new MappingException("Could not write value for property " + persistentProperty.toString(), e);
}
}

36
src/main/java/org/springframework/data/gemfire/mapping/RegionRef.java Normal file
View File

@@ -0,0 +1,36 @@
/*
* 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.mapping;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.data.annotation.Reference;
/**
* @author David Turanski
*
*/
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.FIELD)
@Documented
@Reference
public @interface RegionRef {
String value() default "";
}

12
src/main/java/org/springframework/data/gemfire/mapping/Regions.java
View File

@@ -15,6 +15,7 @@
*/
package org.springframework.data.gemfire.mapping;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
@@ -22,7 +23,6 @@ import java.util.Map;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.util.Assert;
import com.gemstone.bp.edu.emory.mathcs.backport.java.util.Collections;
import com.gemstone.gemfire.cache.Region;
/**
@@ -33,10 +33,12 @@ import com.gemstone.gemfire.cache.Region;
public class Regions implements Iterable<Region<?, ?>> {
private final Map<String, Region<?, ?>> regions;
private final MappingContext<? extends GemfirePersistentEntity<?>, ?> context;
/**
* Creates a new {@link Regions} wrapper for the given {@link Region}s and {@link MappingContext}.
* Creates a new {@link Regions} wrapper for the given {@link Region}s and
* {@link MappingContext}.
*
* @param regions must not be {@literal null}.
* @param context must not be {@literal null}.
@@ -58,8 +60,9 @@ public class Regions implements Iterable<Region<?, ?>> {
}
/**
* Returns the {@link Region} the given type is mapped to. Will try to find a {@link Region} with the simple class
* name in case no mapping information is found.
* Returns the {@link Region} the given type is mapped to. Will try to find
* a {@link Region} with the simple class name in case no mapping
* information is found.
*
* @param type must not be {@literal null}.
* @return the {@link Region} the given type is mapped to.
@@ -89,6 +92,7 @@ public class Regions implements Iterable<Region<?, ?>> {
/*
* (non-Javadoc)
*
* @see java.lang.Iterable#iterator()
*/
@Override

44
src/main/java/org/springframework/data/gemfire/repository/support/GemfireRepositoryFactory.java
View File

@@ -40,13 +40,15 @@ import org.springframework.util.Assert;
import com.gemstone.gemfire.cache.Region;
/**
* {@link RepositoryFactorySupport} implementation creating repository proxies for Gemfire.
* {@link RepositoryFactorySupport} implementation creating repository proxies
* for Gemfire.
*
* @author Oliver Gierke
*/
public class GemfireRepositoryFactory extends RepositoryFactorySupport {
private final MappingContext<? extends GemfirePersistentEntity<?>, GemfirePersistentProperty> context;
private final Regions regions;
/**
@@ -66,7 +68,10 @@ public class GemfireRepositoryFactory extends RepositoryFactorySupport {
/*
* (non-Javadoc)
* @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getEntityInformation(java.lang.Class)
*
* @see
* org.springframework.data.repository.core.support.RepositoryFactorySupport
* #getEntityInformation(java.lang.Class)
*/
@Override
@SuppressWarnings("unchecked")
@@ -78,7 +83,11 @@ public class GemfireRepositoryFactory extends RepositoryFactorySupport {
/*
* (non-Javadoc)
* @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getTargetRepository(org.springframework.data.repository.core.RepositoryMetadata)
*
* @see
* org.springframework.data.repository.core.support.RepositoryFactorySupport
* #getTargetRepository(org.springframework.data.repository.core.
* RepositoryMetadata)
*/
@Override
@SuppressWarnings({ "rawtypes", "unchecked" })
@@ -98,17 +107,21 @@ public class GemfireRepositoryFactory extends RepositoryFactorySupport {
Region<?, ?> region = regions.getRegion(domainClass);
if (region == null) {
throw new IllegalStateException(String.format("No region '%s' found for domain class %s! Make sure you have "
+ "configured a Gemfire region of that name in your application context!", entity.getRegionName(), domainClass));
throw new IllegalStateException(String.format(
"No region '%s' found for domain class %s! Make sure you have "
+ "configured a Gemfire region of that name in your application context!",
entity.getRegionName(), domainClass));
}
Class<?> regionKeyType = region.getAttributes().getKeyConstraint();
Class<?> entityIdType = metadata.getIdType();
if (regionKeyType != null && entity.getIdProperty() != null) {
Assert.isTrue(regionKeyType.isAssignableFrom(entityIdType), String.format(
"The region referenced only supports keys of type %s but the entity to be stored has an id of type %s!",
regionKeyType, entityIdType));
Assert.isTrue(
regionKeyType.isAssignableFrom(entityIdType),
String.format(
"The region referenced only supports keys of type %s but the entity to be stored has an id of type %s!",
regionKeyType, entityIdType));
}
return new GemfireTemplate(region);
@@ -116,7 +129,11 @@ public class GemfireRepositoryFactory extends RepositoryFactorySupport {
/*
* (non-Javadoc)
* @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getRepositoryBaseClass(org.springframework.data.repository.core.RepositoryMetadata)
*
* @see
* org.springframework.data.repository.core.support.RepositoryFactorySupport
* #getRepositoryBaseClass(org.springframework.data.repository.core.
* RepositoryMetadata)
*/
@Override
protected Class<?> getRepositoryBaseClass(RepositoryMetadata metadata) {
@@ -125,7 +142,11 @@ public class GemfireRepositoryFactory extends RepositoryFactorySupport {
/*
* (non-Javadoc)
* @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getQueryLookupStrategy(org.springframework.data.repository.query.QueryLookupStrategy.Key)
*
* @see
* org.springframework.data.repository.core.support.RepositoryFactorySupport
* #getQueryLookupStrategy(org.springframework.data.repository.query.
* QueryLookupStrategy.Key)
*/
@Override
protected QueryLookupStrategy getQueryLookupStrategy(Key key) {
@@ -142,7 +163,8 @@ public class GemfireRepositoryFactory extends RepositoryFactorySupport {
String namedQueryName = queryMethod.getNamedQueryName();
if (namedQueries.hasQuery(namedQueryName)) {
return new StringBasedGemfireRepositoryQuery(namedQueries.getQuery(namedQueryName), queryMethod, template);
return new StringBasedGemfireRepositoryQuery(namedQueries.getQuery(namedQueryName), queryMethod,
template);
}
return new PartTreeGemfireRepositoryQuery(queryMethod, template);

20
src/main/java/org/springframework/data/gemfire/repository/support/GemfireRepositoryFactoryBean.java
View File

@@ -17,6 +17,7 @@ package org.springframework.data.gemfire.repository.support;
import java.io.Serializable;
import java.util.Collection;
import java.util.Collections;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.FactoryBean;
@@ -29,7 +30,6 @@ import org.springframework.data.repository.Repository;
import org.springframework.data.repository.core.support.RepositoryFactoryBeanSupport;
import org.springframework.data.repository.core.support.RepositoryFactorySupport;
import com.gemstone.bp.edu.emory.mathcs.backport.java.util.Collections;
import com.gemstone.gemfire.cache.Region;
/**
@@ -38,20 +38,24 @@ import com.gemstone.gemfire.cache.Region;
* @author Oliver Gierke
*/
public class GemfireRepositoryFactoryBean<T extends Repository<S, ID>, S, ID extends Serializable> extends
RepositoryFactoryBeanSupport<T, S, ID> implements ApplicationContextAware {
RepositoryFactoryBeanSupport<T, S, ID> implements ApplicationContextAware {
private MappingContext<? extends GemfirePersistentEntity<?>, GemfirePersistentProperty> context;
private Iterable<Region<?, ?>> regions;
/*
* (non-Javadoc)
* @see org.springframework.context.ApplicationContextAware#setApplicationContext(org.springframework.context.ApplicationContext)
*
* @see
* org.springframework.context.ApplicationContextAware#setApplicationContext
* (org.springframework.context.ApplicationContext)
*/
@Override
@SuppressWarnings({ "unchecked", "rawtypes" })
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
Collection<Region> regions = applicationContext.getBeansOfType(Region.class).values();
this.regions = Collections.unmodifiableCollection(regions);
this.regions = (Iterable) Collections.unmodifiableCollection(regions);
}
/**
@@ -59,13 +63,17 @@ RepositoryFactoryBeanSupport<T, S, ID> implements ApplicationContextAware {
*
* @param context the context to set
*/
public void setMappingContext(MappingContext<? extends GemfirePersistentEntity<?>, GemfirePersistentProperty> context) {
public void setMappingContext(
MappingContext<? extends GemfirePersistentEntity<?>, GemfirePersistentProperty> context) {
this.context = context;
}
/*
* (non-Javadoc)
* @see org.springframework.data.repository.core.support.RepositoryFactoryBeanSupport#createRepositoryFactory()
*
* @see
* org.springframework.data.repository.core.support.RepositoryFactoryBeanSupport
* #createRepositoryFactory()
*/
@Override
protected RepositoryFactorySupport createRepositoryFactory() {

60
src/main/java/org/springframework/data/gemfire/repository/support/SimpleGemfireRepository.java
View File

@@ -24,6 +24,7 @@ import com.gemstone.gemfire.cache.Region;
public class SimpleGemfireRepository<T, ID extends Serializable> implements GemfireRepository<T, ID> {
private final GemfireTemplate template;
private final EntityInformation<T, ID> entityInformation;
/**
@@ -43,8 +44,10 @@ public class SimpleGemfireRepository<T, ID extends Serializable> implements Gemf
/*
* (non-Javadoc)
*
* @see org.springframework.data.repository.CrudRepository#save(S)
*/
@Override
public <U extends T> U save(U entity) {
template.put(entityInformation.getId(entity), entity);
return entity;
@@ -52,8 +55,12 @@ public class SimpleGemfireRepository<T, ID extends Serializable> implements Gemf
/*
* (non-Javadoc)
* @see org.springframework.data.repository.CrudRepository#save(java.lang.Iterable)
*
* @see
* org.springframework.data.repository.CrudRepository#save(java.lang.Iterable
* )
*/
@Override
public <U extends T> Iterable<U> save(Iterable<U> entities) {
List<U> result = new ArrayList<U>();
for (U entity : entities) {
@@ -64,8 +71,11 @@ public class SimpleGemfireRepository<T, ID extends Serializable> implements Gemf
/*
* (non-Javadoc)
* @see org.springframework.data.repository.CrudRepository#findOne(java.io.Serializable)
*
* @see org.springframework.data.repository.CrudRepository#findOne(java.io.
* Serializable)
*/
@Override
@SuppressWarnings("unchecked")
public T findOne(ID id) {
Object object = template.get(id);
@@ -74,18 +84,24 @@ public class SimpleGemfireRepository<T, ID extends Serializable> implements Gemf
/*
* (non-Javadoc)
* @see org.springframework.data.repository.CrudRepository#exists(java.io.Serializable)
*
* @see org.springframework.data.repository.CrudRepository#exists(java.io.
* Serializable)
*/
@Override
public boolean exists(ID id) {
return findOne(id) != null;
}
/*
* (non-Javadoc)
*
* @see org.springframework.data.repository.CrudRepository#findAll()
*/
@Override
public Collection<T> findAll() {
return template.execute(new GemfireCallback<Collection<T>>() {
@Override
@SuppressWarnings({ "rawtypes", "unchecked" })
public Collection<T> doInGemfire(Region region) {
return region.values();
@@ -93,9 +109,12 @@ public class SimpleGemfireRepository<T, ID extends Serializable> implements Gemf
});
}
/*
/*
* (non-Javadoc)
* @see org.springframework.data.repository.CrudRepository#findAll(java.lang.Iterable)
*
* @see
* org.springframework.data.repository.CrudRepository#findAll(java.lang.
* Iterable)
*/
@Override
@SuppressWarnings("unchecked")
@@ -111,11 +130,15 @@ public class SimpleGemfireRepository<T, ID extends Serializable> implements Gemf
/*
* (non-Javadoc)
*
* @see org.springframework.data.repository.CrudRepository#count()
*/
@Override
public long count() {
return template.execute(new GemfireCallback<Long>() {
@Override
@SuppressWarnings("rawtypes")
// TODO: Take into account keySetOnServer()
public Long doInGemfire(Region region) throws GemFireCheckedException, GemFireException {
return Long.valueOf(region.keySet().size());
}
@@ -124,16 +147,24 @@ public class SimpleGemfireRepository<T, ID extends Serializable> implements Gemf
/*
* (non-Javadoc)
* @see org.springframework.data.repository.CrudRepository#delete(java.lang.Object)
*
* @see
* org.springframework.data.repository.CrudRepository#delete(java.lang.Object
* )
*/
@Override
public void delete(T entity) {
template.remove(entityInformation.getId(entity));
}
/*
* (non-Javadoc)
* @see org.springframework.data.repository.CrudRepository#delete(java.lang.Iterable)
*
* @see
* org.springframework.data.repository.CrudRepository#delete(java.lang.Iterable
* )
*/
@Override
public void delete(Iterable<? extends T> entities) {
for (T entity : entities) {
delete(entity);
@@ -142,11 +173,14 @@ public class SimpleGemfireRepository<T, ID extends Serializable> implements Gemf
/*
* (non-Javadoc)
*
* @see org.springframework.data.repository.CrudRepository#deleteAll()
*/
@Override
public void deleteAll() {
template.execute(new GemfireCallback<Void>() {
@Override
@SuppressWarnings("rawtypes")
public Void doInGemfire(Region region) {
region.clear();
@@ -157,15 +191,21 @@ public class SimpleGemfireRepository<T, ID extends Serializable> implements Gemf
/*
* (non-Javadoc)
* @see org.springframework.data.repository.CrudRepository#delete(java.io.Serializable)
*
* @see org.springframework.data.repository.CrudRepository#delete(java.io.
* Serializable)
*/
@Override
public void delete(ID id) {
template.remove(id);
}
/*
/*
* (non-Javadoc)
* @see org.springframework.data.gemfire.repository.GemfireRepository#save(org.springframework.data.gemfire.repository.Wrapper)
*
* @see
* org.springframework.data.gemfire.repository.GemfireRepository#save(org
* .springframework.data.gemfire.repository.Wrapper)
*/
@Override
public T save(Wrapper<T, ID> wrapper) {

74
src/main/java/org/springframework/data/gemfire/wan/AbstractWANComponentFactoryBean.java Normal file
View File

@@ -0,0 +1,74 @@
/*
* 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.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
import com.gemstone.gemfire.cache.Cache;
/**
* Base class for Gemfire WAN Gateway component factory beans
* @author David Turanski
*
*/
public abstract class AbstractWANComponentFactoryBean<T> implements FactoryBean<T>, InitializingBean, BeanNameAware,
DisposableBean {
protected Log log = LogFactory.getLog(this.getClass());
protected String name;
protected final Cache cache;
protected AbstractWANComponentFactoryBean(Cache cache) {
this.cache = cache;
}
@Override
public void destroy() throws Exception {
// TODO Auto-generated method stub
}
@Override
public final void setBeanName(String name) {
this.name = name;
}
@Override
public final void afterPropertiesSet() throws Exception {
Assert.notNull(name, "Name cannot be null");
Assert.notNull(cache, "Cache cannot be null");
doInit();
}
protected abstract void doInit();
@Override
public abstract T getObject() throws Exception;
@Override
public abstract Class<?> getObjectType();
@Override
public final boolean isSingleton() {
return true;
}
}

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

@@ -0,0 +1,126 @@
/*
* 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.AsyncEventQueue;
import com.gemstone.gemfire.cache.AsyncEventQueueFactory;
import com.gemstone.gemfire.cache.Cache;
import com.gemstone.gemfire.cache.CacheClosedException;
import com.gemstone.gemfire.cache.wan.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 Boolean parallel;
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 = cache.createAsyncEventQueueFactory();
if (parallel != null) {
asyncEventQueueFactory.setParallel(parallel);
}
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 {
asyncEventQueue.stop();
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;
}
public void setParallel(Boolean parallel) {
this.parallel = parallel;
}
}

190
src/main/java/org/springframework/data/gemfire/wan/GatewayHubFactoryBean.java Normal file
View File

@@ -0,0 +1,190 @@
/*
* 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.data.gemfire.wan.GatewayProxy.GatewayQueue;
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.util.Gateway.OrderPolicy;
import com.gemstone.gemfire.cache.util.GatewayEventListener;
import com.gemstone.gemfire.cache.util.GatewayHub;
import com.gemstone.gemfire.cache.util.GatewayQueueAttributes;
/**
* FactoryBean for creating a GemFire {@link GatewayHub} (deprecated in Gemfire
* 7)
* @author David Turanski
*
*/
public class GatewayHubFactoryBean extends AbstractWANComponentFactoryBean<GatewayHub> {
private static List<String> validStartupPolicyValues = Arrays.asList("none", "primary", "secondary");
private static List<String> validOrderPolicyValues = Arrays.asList("KEY,PARTITION,THREAD");
private GatewayHub gatewayHub;
private Integer port;
private String bindAddress;
private Integer maximumTimeBetweenPings;
private Integer socketBufferSize;
private String startupPolicy;
private Boolean manualStart;
private List<GatewayProxy> gateways;
/**
* @param cache the Gemfire cache
*/
public GatewayHubFactoryBean(Cache cache) {
super(cache);
}
@Override
public GatewayHub getObject() throws Exception {
return gatewayHub;
}
@Override
public Class<?> getObjectType() {
return GatewayHub.class;
}
@Override
protected void doInit() {
gatewayHub = cache.addGatewayHub(name, port == null ? GatewayHub.DEFAULT_PORT : port);
if (log.isDebugEnabled()) {
log.debug("added gateway hub " + name);
}
Assert.notNull(cache.getGatewayHub(name));
if (bindAddress != null) {
gatewayHub.setBindAddress(bindAddress);
}
if (manualStart != null) {
gatewayHub.setManualStart(manualStart);
}
if (socketBufferSize != null) {
gatewayHub.setSocketBufferSize(socketBufferSize);
}
if (startupPolicy != null) {
Assert.isTrue(validStartupPolicyValues.contains(startupPolicy), "The value of startup policy:'"
+ startupPolicy + "' is invalid");
gatewayHub.setStartupPolicy(startupPolicy);
}
if (maximumTimeBetweenPings != null) {
gatewayHub.setMaximumTimeBetweenPings(maximumTimeBetweenPings);
}
for (GatewayProxy gateway : gateways) {
Gateway gw = gatewayHub.addGateway(
gateway.getId(),
gateway.getConcurrencyLevel() == null ? Gateway.DEFAULT_CONCURRENCY_LEVEL : gateway
.getConcurrencyLevel());
if (!CollectionUtils.isEmpty(gateway.getEndpoints())) {
for (GatewayProxy.GatewayEndpoint endpoint : gateway.getEndpoints()) {
gw.addEndpoint(endpoint.getId(), endpoint.getHost(), endpoint.getPort());
}
}
if (!CollectionUtils.isEmpty(gateway.getListeners())) {
for (GatewayEventListener listener : gateway.getListeners()) {
gw.addListener(listener);
}
}
if (gateway.getOrderPolicy() != null) {
Assert.isTrue(validOrderPolicyValues.contains(gateway.getOrderPolicy()), "The value of order policy:'"
+ gateway.getOrderPolicy() + "' is invalid");
gw.setOrderPolicy(OrderPolicy.valueOf(gateway.getOrderPolicy()));
}
if (gateway.getSocketBufferSize() != null) {
gw.setSocketBufferSize(gateway.getSocketBufferSize());
}
if (gateway.getSocketReadTimeout() != null) {
gw.setSocketReadTimeout(gateway.getSocketReadTimeout());
}
if (gateway.getQueue() != null) {
GatewayQueue queue = gateway.getQueue();
GatewayQueueAttributes queueAttributes = gw.getQueueAttributes();
if (queue.getAlertThreshold() != null) {
queueAttributes.setAlertThreshold(queue.getAlertThreshold());
}
if (queue.getEnableBatchConflation() != null) {
queueAttributes.setBatchConflation(queue.getEnableBatchConflation());
}
if (queue.getBatchSize() != null) {
queueAttributes.setBatchSize(queue.getBatchSize());
}
if (queue.getBatchTimeInterval() != null) {
queueAttributes.setBatchTimeInterval(queue.getBatchTimeInterval());
}
if (queue.getDiskStoreRef() != null) {
boolean persistent = (queue.getPersistent() == null) ? Boolean.TRUE : queue.getPersistent();
Assert.isTrue(persistent, "specifying a disk store requires persistent property to be true");
queueAttributes.setDiskStoreName(queue.getDiskStoreRef());
}
if (queue.getPersistent() != null) {
queueAttributes.setEnablePersistence(queue.getPersistent());
}
if (queue.getMaximumQueueMemory() != null) {
queueAttributes.setMaximumQueueMemory(queue.getMaximumQueueMemory());
}
}
}
}
public void setPort(Integer port) {
this.port = port;
}
public void setBindAddress(String bindAddress) {
this.bindAddress = bindAddress;
}
public void setMaximumTimeBetweenPings(Integer maximumTimeBetweenPings) {
this.maximumTimeBetweenPings = maximumTimeBetweenPings;
}
public void setSocketBufferSize(Integer socketBufferSize) {
this.socketBufferSize = socketBufferSize;
}
public void setStartupPolicy(String startupPolicy) {
this.startupPolicy = startupPolicy;
}
public void setManualStart(Boolean manualStart) {
this.manualStart = manualStart;
}
public void setGateways(List<GatewayProxy> gateways) {
this.gateways = gateways;
}
}

212
src/main/java/org/springframework/data/gemfire/wan/GatewayProxy.java Normal file
View File

@@ -0,0 +1,212 @@
/*
* 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 com.gemstone.gemfire.cache.util.GatewayEventListener;
/**
* This class used to allow decoupling of 'gateway' parsing from 'gateway-hub'
* parsing
*
* @author David Turanski
*
*/
public class GatewayProxy {
private List<GatewayEndpoint> endpoints;
private Integer concurrencyLevel;
private String id;
private List<GatewayEventListener> listeners;
private String orderPolicy;
private int socketBufferSize;
private int socketReadTimeout;
private GatewayQueue queue;
public void setEndpoints(List<GatewayEndpoint> endpoints) {
this.endpoints = endpoints;
}
public void setListeners(List<GatewayEventListener> listeners) {
this.listeners = listeners;
}
public void setQueue(GatewayQueue queue) {
this.queue = queue;
}
public GatewayQueue getQueue() {
return this.queue;
}
public Integer getConcurrencyLevel() {
return this.concurrencyLevel;
}
public List<GatewayEndpoint> getEndpoints() {
return endpoints;
}
public String getId() {
return this.id;
}
public List<GatewayEventListener> getListeners() {
return this.listeners;
}
public String getOrderPolicy() {
return this.orderPolicy;
}
public Integer getSocketBufferSize() {
return this.socketBufferSize;
}
public Integer getSocketReadTimeout() {
return this.socketReadTimeout;
}
public void setId(String id) {
this.id = id;
}
public void setOrderPolicy(String orderPolicy) {
this.orderPolicy = orderPolicy;
}
public void setSocketBufferSize(int socketBufferSize) {
this.socketBufferSize = socketBufferSize;
}
public void setSocketReadTimeout(int socketReadTimeout) {
this.socketReadTimeout = socketReadTimeout;
}
public static class GatewayEndpoint {
private String host;
private String id;
private int port;
public String getHost() {
return host;
}
public void setHost(String host) {
this.host = host;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public int getPort() {
return port;
}
public void setPort(int port) {
this.port = port;
}
}
public static class GatewayQueue {
private Integer alertThreshold;
private Boolean enableBatchConflation;
private Integer batchTimeInterval;
private Integer batchSize;
private Boolean persistent;
private String diskStoreRef;
private Integer maximumQueueMemory;
public Integer getAlertThreshold() {
return alertThreshold;
}
public void setAlertThreshold(Integer alertThreshold) {
this.alertThreshold = alertThreshold;
}
public Boolean getEnableBatchConflation() {
return enableBatchConflation;
}
public void setEnableBatchConflation(Boolean enableBatchConflation) {
this.enableBatchConflation = enableBatchConflation;
}
public Integer getBatchTimeInterval() {
return batchTimeInterval;
}
public void setBatchTimeInterval(Integer batchTimeInterval) {
this.batchTimeInterval = batchTimeInterval;
}
public Integer getBatchSize() {
return batchSize;
}
public void setBatchSize(Integer batchSize) {
this.batchSize = batchSize;
}
public Boolean getPersistent() {
return persistent;
}
public void setPersistent(Boolean persistent) {
this.persistent = persistent;
}
public String getDiskStoreRef() {
return diskStoreRef;
}
public void setDiskStoreRef(String diskStoreRef) {
this.diskStoreRef = diskStoreRef;
}
public Integer getMaximumQueueMemory() {
return maximumQueueMemory;
}
public void setMaximumQueueMemory(Integer maximumQueueMemory) {
this.maximumQueueMemory = maximumQueueMemory;
}
}
}

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

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

@@ -0,0 +1,236 @@
/*
* 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 = cache.createGatewaySenderFactory();
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 setGatewaySender(GatewaySender gatewaySender) {
this.gatewaySender = gatewaySender;
}
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;
}
}