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

Implements JIRA feature request SGF-227 adding support for GemFire 8.0's auto-reconnect functionality on forced disconnects.

Browse Source
This commit is contained in:
John Blum
2014-07-09 23:03:14 -07:00
parent d8896fd05b
commit 76aa819be7
12 changed files with 607 additions and 330 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -89,6 +89,7 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
protected BeanFactory beanFactory;
protected Boolean copyOnRead;
protected Boolean enableAutoReconnect;
protected Boolean pdxIgnoreUnreadFields;
protected Boolean pdxPersistent;
protected Boolean pdxReadSerialized;
@@ -258,16 +259,14 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
catch (CacheClosedException ex) {
initializeDynamicRegionFactory();
Object factory = createFactory(this.properties);
Object factory = createFactory(getProperties());
// GemFire 6.6 specific options
if (isPdxSettingsSpecified()) {
Assert.isTrue(ClassUtils.isPresent("com.gemstone.gemfire.pdx.PdxSerializer", beanClassLoader),
"Cannot set PDX options since GemFire 6.6 not detected.");
applyPdxOptions(factory);
}
// fall back to cache creation
cache = (Cache) createCache(factory);
messagePrefix = "Created new";
}
@@ -275,20 +274,20 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
if (this.copyOnRead != null) {
cache.setCopyOnRead(this.copyOnRead);
}
if (gatewayConflictResolver != null) {
cache.setGatewayConflictResolver((GatewayConflictResolver) gatewayConflictResolver);
}
if (lockLease != null) {
cache.setLockLease(lockLease);
}
if (lockTimeout != null) {
cache.setLockTimeout(lockTimeout);
}
if (searchTimeout != null) {
cache.setSearchTimeout(searchTimeout);
}
if (messageSyncInterval != null) {
cache.setMessageSyncInterval(messageSyncInterval);
}
if (gatewayConflictResolver != null) {
cache.setGatewayConflictResolver((GatewayConflictResolver) gatewayConflictResolver);
if (searchTimeout != null) {
cache.setSearchTimeout(searchTimeout);
}
DistributedSystem system = cache.getDistributedSystem();
@@ -306,7 +305,7 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
cache.loadCacheXml(cacheXml.getInputStream());
if (log.isDebugEnabled()) {
log.debug("Initialized cache from " + cacheXml);
log.debug(String.format("Initialized Cache from '%1$s'.", cacheXml));
}
}
@@ -404,12 +403,12 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
return (cache != null ? cache : ((CacheFactory) factory).create());
}
protected GemFireCache fetchCache() {
return (cache != null ? cache : CacheFactory.getAnyInstance());
protected Object createFactory(Properties gemfireProperties) {
return new CacheFactory(gemfireProperties);
}
protected Object createFactory(Properties props) {
return new CacheFactory(props);
protected GemFireCache fetchCache() {
return (cache != null ? cache : CacheFactory.getAnyInstance());
}
@Override
@@ -481,6 +480,15 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
this.beanName = name;
}
/**
* Sets the cache configuration.
*
* @param cacheXml the cacheXml to set
*/
public void setCacheXml(Resource cacheXml) {
this.cacheXml = cacheXml;
}
/**
* Sets the cache properties.
*
@@ -491,12 +499,10 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
}
/**
* Sets the cache configuration.
*
* @param cacheXml the cacheXml to set
* @param lazyInitialize set to false to force cache initialization if no other bean references it
*/
public void setCacheXml(Resource cacheXml) {
this.cacheXml = cacheXml;
public void setLazyInitialize(boolean lazyInitialize) {
this.lazyInitialize = lazyInitialize;
}
/**
@@ -512,6 +518,10 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
this.useBeanFactoryLocator = usage;
}
public void setEnableAutoReconnect(final Boolean enableAutoReconnect) {
this.enableAutoReconnect = enableAutoReconnect;
}
/**
* Sets the {@link PdxSerializable} for this cache. Applicable on GemFire
* 6.6 or higher. The argument is of type object for compatibility with
@@ -701,10 +711,17 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
}
/**
* @param lazyInitialize set to false to force cache initialization if no other bean references it
* @return the beanClassLoader
*/
public void setLazyInitialize(boolean lazyInitialize) {
this.lazyInitialize = lazyInitialize;
public ClassLoader getBeanClassLoader() {
return beanClassLoader;
}
/**
* @return the beanName
*/
public String getBeanName() {
return beanName;
}
/**
@@ -718,21 +735,15 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
* @return the properties
*/
public Properties getProperties() {
if (properties == null) {
properties = new Properties();
}
return properties;
}
/**
* @return the beanClassLoader
*/
public ClassLoader getBeanClassLoader() {
return beanClassLoader;
}
/**
* @return the beanName
*/
public String getBeanName() {
return beanName;
public Boolean getEnableAutoReconnect() {
return enableAutoReconnect;
}
/**
@@ -864,12 +875,19 @@ public class CacheFactoryBean implements BeanNameAware, BeanFactoryAware, BeanCl
return lazyInitialize;
}
protected void postProcessPropertiesBeforeInitialization(Properties gemfireProperties) {
gemfireProperties.setProperty("disable-auto-reconnect", String.valueOf(
!Boolean.TRUE.equals(getEnableAutoReconnect())));
}
/* (non-Javadoc)
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
@Override
public void afterPropertiesSet() throws Exception {
if (!lazyInitialize) {
postProcessPropertiesBeforeInitialization(getProperties());
if (!isLazyInitialize()) {
init();
}
}

59
src/main/java/org/springframework/data/gemfire/client/ClientCacheFactoryBean.java
View File

@@ -86,11 +86,12 @@ public class ClientCacheFactoryBean extends CacheFactoryBean {
@Override
protected GemFireCache createCache(Object factory) {
ClientCacheFactory ccf = (ClientCacheFactory) factory;
initializePool(ccf);
ClientCacheFactory clientCacheFactory = (ClientCacheFactory) factory;
initializePool(clientCacheFactory);
// Now create the cache
GemFireCache cache = ccf.create();
GemFireCache cache = clientCacheFactory.create();
// Register for events after pool/regions been created and iff non-durable client
readyForEvents();
@@ -100,8 +101,8 @@ public class ClientCacheFactoryBean extends CacheFactoryBean {
}
@Override
protected Object createFactory(Properties props) {
return new ClientCacheFactory(props);
protected Object createFactory(Properties gemfireProperties) {
return new ClientCacheFactory(gemfireProperties);
}
@Override
@@ -109,10 +110,6 @@ public class ClientCacheFactoryBean extends CacheFactoryBean {
return ClientCacheFactory.getAnyInstance();
}
public Properties getProperties() {
return this.properties;
}
private void initializePool(ClientCacheFactory ccf) {
Pool p = pool;
@@ -189,20 +186,18 @@ public class ClientCacheFactoryBean extends CacheFactoryBean {
}
}
/**
* Inform the GemFire cluster that this client cache is ready to receive events.
*/
private void readyForEvents(){
ClientCache clientCache = ClientCacheFactory.getAnyInstance();
@Override
protected void postProcessPropertiesBeforeInitialization(final Properties gemfireProperties) {
}
if (Boolean.TRUE.equals(readyForEvents) && !clientCache.isClosed()) {
try {
clientCache.readyForEvents();
}
catch (IllegalStateException ignore) {
// Cannot be called for a non-durable client so exception is thrown.
}
}
@Override
public final Boolean getEnableAutoReconnect() {
return Boolean.FALSE;
}
@Override
public final void setEnableAutoReconnect(final Boolean enableAutoReconnect) {
throw new UnsupportedOperationException("Auto-reconnect is not supported on ClientCache.");
}
/**
@@ -235,11 +230,11 @@ public class ClientCacheFactoryBean extends CacheFactoryBean {
public void setReadyForEvents(Boolean readyForEvents){
this.readyForEvents = readyForEvents;
}
public Boolean getReadyForEvents(){
return this.readyForEvents;
}
@Override
protected void applyPdxOptions(Object factory) {
if (factory instanceof ClientCacheFactory) {
@@ -247,4 +242,20 @@ public class ClientCacheFactoryBean extends CacheFactoryBean {
}
}
/**
* Inform the GemFire cluster that this client cache is ready to receive events.
*/
private void readyForEvents(){
ClientCache clientCache = ClientCacheFactory.getAnyInstance();
if (Boolean.TRUE.equals(readyForEvents) && !clientCache.isClosed()) {
try {
clientCache.readyForEvents();
}
catch (IllegalStateException ignore) {
// cannot be called for a non-durable client so exception is thrown
}
}
}
}

91
src/main/java/org/springframework/data/gemfire/config/CacheParser.java
View File

@@ -56,49 +56,54 @@ class CacheParser extends AbstractSimpleBeanDefinitionParser {
ParsingUtils.setPropertyValue(element, builder, "cache-xml-location", "cacheXml");
ParsingUtils.setPropertyReference(element, builder, "properties-ref", "properties");
ParsingUtils.setPropertyReference(element, builder, "pdx-serializer-ref", "pdxSerializer");
parsePdxDiskStore(element, parserContext, builder);
ParsingUtils.setPropertyValue(element, builder, "pdx-persistent");
ParsingUtils.setPropertyValue(element, builder, "pdx-read-serialized");
ParsingUtils.setPropertyValue(element, builder, "pdx-ignore-unread-fields");
ParsingUtils.setPropertyValue(element, builder, "use-bean-factory-locator");
ParsingUtils.setPropertyValue(element, builder, "close");
ParsingUtils.setPropertyValue(element, builder, "copy-on-read");
ParsingUtils.setPropertyValue(element, builder, "lock-timeout");
ParsingUtils.setPropertyValue(element, builder, "lock-lease");
ParsingUtils.setPropertyValue(element, builder, "message-sync-interval");
ParsingUtils.setPropertyValue(element, builder, "search-timeout");
ParsingUtils.setPropertyValue(element, builder, "critical-heap-percentage");
ParsingUtils.setPropertyValue(element, builder, "eviction-heap-percentage");
ParsingUtils.setPropertyValue(element, builder, "close");
ParsingUtils.setPropertyValue(element, builder, "enable-auto-reconnect");
ParsingUtils.setPropertyValue(element, builder, "lock-lease");
ParsingUtils.setPropertyValue(element, builder, "lock-timeout");
ParsingUtils.setPropertyValue(element, builder, "message-sync-interval");
ParsingUtils.setPropertyReference(element, builder, "pdx-serializer-ref", "pdxSerializer");
ParsingUtils.setPropertyValue(element, builder, "pdx-read-serialized");
ParsingUtils.setPropertyValue(element, builder, "pdx-ignore-unread-fields");
ParsingUtils.setPropertyValue(element, builder, "pdx-persistent");
parsePdxDiskStore(element, parserContext, builder);
ParsingUtils.setPropertyValue(element, builder, "search-timeout");
ParsingUtils.setPropertyValue(element, builder, "lazy-init","lazyInitialize");
ParsingUtils.setPropertyValue(element, builder, "use-bean-factory-locator");
List<Element> txListeners = DomUtils.getChildElementsByTagName(element, "transaction-listener");
if (!CollectionUtils.isEmpty(txListeners)) {
ManagedList<Object> transactionListeners = new ManagedList<Object>();
for (Element txListener : txListeners) {
transactionListeners.add(ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, txListener,
builder));
transactionListeners.add(ParsingUtils.parseRefOrNestedBeanDeclaration(
parserContext, txListener, builder));
}
builder.addPropertyValue("transactionListeners", transactionListeners);
}
Element txWriter = DomUtils.getChildElementByTagName(element, "transaction-writer");
if (txWriter != null) {
builder.addPropertyValue("transactionWriter",
ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, txWriter, builder));
builder.addPropertyValue("transactionWriter", ParsingUtils.parseRefOrNestedBeanDeclaration(
parserContext, txWriter, builder));
}
Element gatewayConflictResolver = DomUtils.getChildElementByTagName(element, "gateway-conflict-resolver");
if (gatewayConflictResolver != null) {
ParsingUtils.throwExceptionIfNotGemfireV7(element.getLocalName(), "gateway-conflict-resolver",
parserContext);
builder.addPropertyValue("gatewayConflictResolver",
ParsingUtils.parseRefOrSingleNestedBeanDeclaration(parserContext, gatewayConflictResolver, builder));
ParsingUtils.throwExceptionIfNotGemfireV7(element.getLocalName(), "gateway-conflict-resolver", parserContext);
builder.addPropertyValue("gatewayConflictResolver", ParsingUtils.parseRefOrSingleNestedBeanDeclaration(
parserContext, gatewayConflictResolver, builder));
}
Element function = DomUtils.getChildElementByTagName(element, "function");
if (function != null) {
builder.addPropertyValue("functions",
ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, function, builder));
builder.addPropertyValue("functions", ParsingUtils.parseRefOrNestedBeanDeclaration(
parserContext, function, builder));
}
parseDynamicRegionFactory(element, builder);
@@ -130,6 +135,7 @@ class CacheParser extends AbstractSimpleBeanDefinitionParser {
private void parseDynamicRegionFactory(Element element, BeanDefinitionBuilder builder) {
Element dynamicRegionFactory = DomUtils.getChildElementByTagName(element, "dynamic-region-factory");
if (dynamicRegionFactory != null) {
BeanDefinitionBuilder dynamicRegionSupport = buildDynamicRegionSupport(dynamicRegionFactory);
postProcessDynamicRegionSupport(element, dynamicRegionSupport);
@@ -137,35 +143,40 @@ class CacheParser extends AbstractSimpleBeanDefinitionParser {
}
}
/**
* @param dynamicRegionSupport BDB for &lt;dynamic-region-factory&gt;
* element
*/
protected void postProcessDynamicRegionSupport(Element element, BeanDefinitionBuilder dynamicRegionSupport) {
}
private BeanDefinitionBuilder buildDynamicRegionSupport(Element dynamicRegionFactory) {
BeanDefinitionBuilder result = null;
if (dynamicRegionFactory != null) {
BeanDefinitionBuilder dynamicRegionSupport = BeanDefinitionBuilder
.genericBeanDefinition(CacheFactoryBean.DynamicRegionSupport.class);
String diskDir = dynamicRegionFactory.getAttribute("disk-dir");
if (StringUtils.hasText(diskDir)) {
dynamicRegionSupport.addPropertyValue("diskDir", diskDir);
BeanDefinitionBuilder dynamicRegionSupport = BeanDefinitionBuilder.genericBeanDefinition(
CacheFactoryBean.DynamicRegionSupport.class);
String diskDirectory = dynamicRegionFactory.getAttribute("disk-dir");
if (StringUtils.hasText(diskDirectory)) {
dynamicRegionSupport.addPropertyValue("diskDir", diskDirectory);
}
String persistent = dynamicRegionFactory.getAttribute("persistent");
if (StringUtils.hasText(persistent)) {
dynamicRegionSupport.addPropertyValue("persistent", persistent);
}
String registerInterest = dynamicRegionFactory.getAttribute("register-interest");
if (StringUtils.hasText(registerInterest)) {
dynamicRegionSupport.addPropertyValue("registerInterest", registerInterest);
}
result = dynamicRegionSupport;
return dynamicRegionSupport;
}
return result;
return null;
}
/**
* @param dynamicRegionSupport BDB for &lt;dynamic-region-factory&gt;
* element
*/
protected void postProcessDynamicRegionSupport(Element element, BeanDefinitionBuilder dynamicRegionSupport) {
}
private void parseJndiBindings(Element element, BeanDefinitionBuilder builder) {
@@ -222,11 +233,13 @@ class CacheParser extends AbstractSimpleBeanDefinitionParser {
protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
throws BeanDefinitionStoreException {
String name = super.resolveId(element, definition, parserContext);
if (!StringUtils.hasText(name)) {
name = GemfireConstants.DEFAULT_GEMFIRE_CACHE_NAME;
//For backward compatibility
parserContext.getRegistry().registerAlias(GemfireConstants.DEFAULT_GEMFIRE_CACHE_NAME, "gemfire-cache");
// Set Cache bean alias for backwards compatibility...
parserContext.getRegistry().registerAlias(name, "gemfire-cache");
}
return name;
}