From dc91f82daf305e6a87bf7e8afc135ad1b73bc615 Mon Sep 17 00:00:00 2001 From: John Blum Date: Mon, 15 Jul 2019 18:37:37 -0700 Subject: [PATCH] SGF-863 - Add support in the XML Namespace for Region to AsyncEventQueue and GatewaySender associations by ID. --- .../data/gemfire/LookupRegionFactoryBean.java | 60 +++++++++++++++--- .../data/gemfire/PeerRegionFactoryBean.java | 62 ++++++++++++++++--- .../config/xml/AbstractRegionParser.java | 26 ++++---- .../data/gemfire/config/xml/ParsingUtils.java | 7 ++- .../gemfire/config/spring-gemfire-2.1.xsd | 47 ++++++++++++++ .../gemfire/config/spring-gemfire-2.2.xsd | 47 ++++++++++++++ .../xml/AbstractRegionParserUnitTests.java | 54 +++++++++++----- 7 files changed, 258 insertions(+), 45 deletions(-) diff --git a/src/main/java/org/springframework/data/gemfire/LookupRegionFactoryBean.java b/src/main/java/org/springframework/data/gemfire/LookupRegionFactoryBean.java index 8851c481..ec9bb724 100644 --- a/src/main/java/org/springframework/data/gemfire/LookupRegionFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/LookupRegionFactoryBean.java @@ -13,13 +13,15 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.springframework.data.gemfire; import static org.springframework.data.gemfire.util.ArrayUtils.nullSafeArray; import java.util.Arrays; +import java.util.HashSet; import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; import org.apache.geode.cache.CacheListener; import org.apache.geode.cache.CacheLoader; @@ -29,7 +31,9 @@ import org.apache.geode.cache.ExpirationAttributes; import org.apache.geode.cache.Region; import org.apache.geode.cache.asyncqueue.AsyncEventQueue; import org.apache.geode.cache.wan.GatewaySender; + import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * The LookupRegionFactoryBean class is a concrete implementation of ResolvableRegionFactoryBean for handling @@ -66,6 +70,9 @@ public class LookupRegionFactoryBean extends ResolvableRegionFactoryBean extends ResolvableRegionFactoryBean { - Arrays.stream(nullSafeArray(this.asyncEventQueues, AsyncEventQueue.class)) - .map(AsyncEventQueue::getId) - .forEach(attributesMutator::addAsyncEventQueueId); + // AsyncEventQueues (AEQ) + getConfiguredAsyncEventQueueIds().forEach(attributesMutator::addAsyncEventQueueId); + // CacheListeners Arrays.stream(nullSafeArray(this.cacheListeners, CacheListener.class)) .forEach(attributesMutator::addCacheListener); @@ -102,12 +109,43 @@ public class LookupRegionFactoryBean extends ResolvableRegionFactoryBean getConfiguredAsyncEventQueueIds() { + + Set asyncEventQueueIds = new HashSet<>(); + + Arrays.stream(nullSafeArray(this.asyncEventQueues, AsyncEventQueue.class)) + .map(AsyncEventQueue::getId) + .collect(Collectors.toCollection(() -> asyncEventQueueIds)); + + Arrays.stream(nullSafeArray(this.asyncEventQueueIds, String.class)) + .filter(StringUtils::hasText) + .map(String::trim) + .collect(Collectors.toCollection(() -> asyncEventQueueIds)); + + return asyncEventQueueIds; + } + + private Set getConfiguredGatewaySenderIds() { + + Set gatewaySenderIds = new HashSet<>(); + + Arrays.stream(nullSafeArray(this.gatewaySenders, GatewaySender.class)) + .map(GatewaySender::getId) + .collect(Collectors.toCollection(() -> gatewaySenderIds)); + + Arrays.stream(nullSafeArray(this.gatewaySenderIds, String.class)) + .filter(StringUtils::hasText) + .map(String::trim) + .collect(Collectors.toCollection(() -> gatewaySenderIds)); + + return gatewaySenderIds; + } + @Override public final boolean isLookupEnabled() { return true; @@ -117,6 +155,10 @@ public class LookupRegionFactoryBean extends ResolvableRegionFactoryBean[] cacheListeners) { this.cacheListeners = cacheListeners; } @@ -161,6 +203,10 @@ public class LookupRegionFactoryBean extends ResolvableRegionFactoryBean extends ConfigurableRegionFact private String diskStoreName; + private String[] asyncEventQueueIds; + private String[] gatewaySenderIds; + /** * Creates a new {@link Region} with the given {@link String name}. * @@ -252,8 +258,7 @@ public abstract class PeerRegionFactoryBean extends ConfigurableRegionFact regionFactory.setStatisticsEnabled(resolveStatisticsEnabled()); - stream(nullSafeArray(this.asyncEventQueues, AsyncEventQueue.class)) - .forEach(asyncEventQueue -> regionFactory.addAsyncEventQueueId(asyncEventQueue.getId())); + getConfiguredAsyncEventQueueIds().forEach(regionFactory::addAsyncEventQueueId); stream(nullSafeArray(this.cacheListeners, CacheListener.class)).forEach(regionFactory::addCacheListener); @@ -279,8 +284,7 @@ public abstract class PeerRegionFactoryBean extends ConfigurableRegionFact Optional.ofNullable(this.evictionAttributes).ifPresent(regionFactory::setEvictionAttributes); - stream(nullSafeArray(this.gatewaySenders, GatewaySender.class)) - .forEach(gatewaySender -> regionFactory.addGatewaySenderId(gatewaySender.getId())); + getConfiguredGatewaySenderIds().forEach(regionFactory::addGatewaySenderId); Optional.ofNullable(this.keyConstraint).ifPresent(regionFactory::setKeyConstraint); @@ -312,6 +316,38 @@ public abstract class PeerRegionFactoryBean extends ConfigurableRegionFact return regionFactory; } + private Set getConfiguredAsyncEventQueueIds() { + + Set asyncEventQueueIds = new HashSet<>(); + + Arrays.stream(nullSafeArray(this.asyncEventQueues, AsyncEventQueue.class)) + .map(AsyncEventQueue::getId) + .collect(Collectors.toCollection(() -> asyncEventQueueIds)); + + Arrays.stream(nullSafeArray(this.asyncEventQueueIds, String.class)) + .filter(StringUtils::hasText) + .map(String::trim) + .collect(Collectors.toCollection(() -> asyncEventQueueIds)); + + return asyncEventQueueIds; + } + + private Set getConfiguredGatewaySenderIds() { + + Set gatewaySenderIds = new HashSet<>(); + + Arrays.stream(nullSafeArray(this.gatewaySenders, GatewaySender.class)) + .map(GatewaySender::getId) + .collect(Collectors.toCollection(() -> gatewaySenderIds)); + + Arrays.stream(nullSafeArray(this.gatewaySenderIds, String.class)) + .filter(StringUtils::hasText) + .map(String::trim) + .collect(Collectors.toCollection(() -> gatewaySenderIds)); + + return gatewaySenderIds; + } + /* * (non-Javadoc) * @@ -627,6 +663,10 @@ public abstract class PeerRegionFactoryBean extends ConfigurableRegionFact this.asyncEventQueues = asyncEventQueues; } + public void setAsyncEventQueueIds(String[] asyncEventQueueIds) { + this.asyncEventQueueIds = asyncEventQueueIds; + } + /** * Sets the {@link RegionAttributes} used to configure this {@link Region}. * @@ -760,8 +800,10 @@ public abstract class PeerRegionFactoryBean extends ConfigurableRegionFact * @see org.apache.geode.cache.DataPolicy */ public DataPolicy getDataPolicy() { - return Optional.ofNullable(this.dataPolicy) - .orElseThrow(() -> newIllegalStateException("Data Policy has not been properly resolved yet")); + + Assert.state(this.dataPolicy != null, "Data Policy has not been properly resolved yet"); + + return this.dataPolicy; } /** @@ -775,8 +817,6 @@ public abstract class PeerRegionFactoryBean extends ConfigurableRegionFact this.diskStoreName = diskStoreName; } - // TODO: review/add Javadoc from here forward... - public void setEntryIdleTimeout(ExpirationAttributes entryIdleTimeout) { this.entryIdleTimeout = entryIdleTimeout; } @@ -801,6 +841,10 @@ public abstract class PeerRegionFactoryBean extends ConfigurableRegionFact this.gatewaySenders = gatewaySenders; } + public void setGatewaySenderIds(String[] gatewaySenderIds) { + this.gatewaySenderIds = gatewaySenderIds; + } + /** * Configures whether to enable this {@link Region} with the ability to store data in {@literal off-heap memory}. * diff --git a/src/main/java/org/springframework/data/gemfire/config/xml/AbstractRegionParser.java b/src/main/java/org/springframework/data/gemfire/config/xml/AbstractRegionParser.java index 58b27906..d0bd686d 100644 --- a/src/main/java/org/springframework/data/gemfire/config/xml/AbstractRegionParser.java +++ b/src/main/java/org/springframework/data/gemfire/config/xml/AbstractRegionParser.java @@ -13,7 +13,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.springframework.data.gemfire.config.xml; import java.util.HashMap; @@ -23,8 +22,7 @@ import java.util.Optional; import org.apache.geode.cache.asyncqueue.AsyncEventQueue; import org.apache.geode.cache.wan.GatewaySender; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; + import org.springframework.beans.PropertyValue; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; @@ -37,6 +35,9 @@ import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.w3c.dom.Element; /** @@ -107,10 +108,10 @@ abstract class AbstractRegionParser extends AbstractSingleBeanDefinitionParser { mergeRegionTemplateAttributes(element, parserContext, regionBuilder, regionAttributesBuilder); - String resolvedCacheRef = ParsingUtils.resolveCacheReference(element.getAttribute("cache-ref")); + String resolvedCacheReference = ParsingUtils.resolveCacheReference(element.getAttribute("cache-ref")); if (!subRegion) { - regionBuilder.addPropertyReference("cache", resolvedCacheRef); + regionBuilder.addPropertyReference("cache", resolvedCacheReference); ParsingUtils.setPropertyValue(element, regionBuilder, "close"); ParsingUtils.setPropertyValue(element, regionBuilder, "destroy"); } @@ -158,7 +159,7 @@ abstract class AbstractRegionParser extends AbstractSingleBeanDefinitionParser { } if (!subRegion) { - parseSubRegions(element, parserContext, resolvedCacheRef); + parseSubRegions(element, parserContext, resolvedCacheReference); } } @@ -255,7 +256,7 @@ abstract class AbstractRegionParser extends AbstractSingleBeanDefinitionParser { String name = element.getAttribute(NAME_ATTRIBUTE); - return (StringUtils.hasText(name) ? name : element.getAttribute(ID_ATTRIBUTE)); + return StringUtils.hasText(name) ? name : element.getAttribute(ID_ATTRIBUTE); } private String buildSubRegionPath(String parentName, String regionName) { @@ -269,7 +270,6 @@ abstract class AbstractRegionParser extends AbstractSingleBeanDefinitionParser { return regionPath; } - /* (non-Javadoc) */ private BeanDefinition parseSubRegion(Element element, ParserContext parserContext, String subRegionPath, String cacheRef) { @@ -288,7 +288,6 @@ abstract class AbstractRegionParser extends AbstractSingleBeanDefinitionParser { return beanDefinition; } - /* (non-Javadoc) */ private String getParentRegionPathFrom(String regionPath) { int index = regionPath.lastIndexOf("/"); @@ -302,13 +301,14 @@ abstract class AbstractRegionParser extends AbstractSingleBeanDefinitionParser { return parentPath; } - /* (non-Javadoc) */ protected void validateDataPolicyShortcutAttributesMutualExclusion(Element element, ParserContext parserContext) { if (element.hasAttribute("data-policy") && element.hasAttribute("shortcut")) { - parserContext.getReaderContext().error(String.format( - "Only one of [data-policy, shortcut] may be specified with element '%1$s'.", element.getTagName()), - element); + + String message = String.format("Only one of [data-policy, shortcut] may be specified with element [%s]", + element.getTagName()); + + parserContext.getReaderContext().error(message, element); } } } diff --git a/src/main/java/org/springframework/data/gemfire/config/xml/ParsingUtils.java b/src/main/java/org/springframework/data/gemfire/config/xml/ParsingUtils.java index 1378fc29..8cd4e7a5 100644 --- a/src/main/java/org/springframework/data/gemfire/config/xml/ParsingUtils.java +++ b/src/main/java/org/springframework/data/gemfire/config/xml/ParsingUtils.java @@ -123,6 +123,7 @@ abstract class ParsingUtils { static Object getBeanReference(Element element, ParserContext parserContext, String refAttributeName) { String refAttributeValue = element.getAttribute(refAttributeName); + Object returnValue = null; if (StringUtils.hasText(refAttributeValue)) { if (!DomUtils.getChildElements(element).isEmpty()) { @@ -134,10 +135,10 @@ abstract class ParsingUtils { parserContext.getReaderContext().error(message, element); } - return null; + returnValue = new RuntimeBeanReference(refAttributeValue); } - return new RuntimeBeanReference(refAttributeValue); + return returnValue; } static Object parseRefOrNestedCustomElement(Element element, ParserContext parserContext, @@ -377,11 +378,13 @@ abstract class ParsingUtils { static void parseOptionalRegionAttributes(Element element, ParserContext parserContext, BeanDefinitionBuilder regionAttributesBuilder) { + setPropertyValue(element, regionAttributesBuilder, "async-event-queue-ids"); setPropertyValue(element, regionAttributesBuilder, "cloning-enabled"); setPropertyValue(element, regionAttributesBuilder, "concurrency-level"); setPropertyValue(element, regionAttributesBuilder, "disk-synchronous"); setPropertyValue(element, regionAttributesBuilder, "enable-async-conflation"); setPropertyValue(element, regionAttributesBuilder, "enable-subscription-conflation"); + setPropertyValue(element, regionAttributesBuilder, "gateway-sender-ids"); setPropertyValue(element, regionAttributesBuilder, "ignore-jta", "ignoreJTA"); setPropertyValue(element, regionAttributesBuilder, "index-update-type"); setPropertyValue(element, regionAttributesBuilder, "initial-capacity"); diff --git a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-2.1.xsd b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-2.1.xsd index 3be167e6..90b267cc 100644 --- a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-2.1.xsd +++ b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-2.1.xsd @@ -626,6 +626,15 @@ use inner bean declarations. + + + + + + + + + + @@ -1050,6 +1069,34 @@ use inner bean declarations. + + + + + + + + + + + + + + + + + + + + + + + + + @@ -1050,6 +1069,34 @@ use inner bean declarations. + + + + + + + + + + + + + + +