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

DATAGEODE-242 - Add support for AEQ pauseEventDispatching.

Browse Source
This commit is contained in:
John Blum
2019-10-27 23:27:05 -07:00
parent 24ad51f7b3
commit 31d08b79fd
11 changed files with 302 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/main/java/org/springframework/data/gemfire/config/xml/AsyncEventQueueParser.java
View File

@@ -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 org.w3c.dom.Element;
@@ -21,6 +20,7 @@ import org.w3c.dom.Element;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.BeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.data.gemfire.util.SpringUtils;
import org.springframework.data.gemfire.wan.AsyncEventQueueFactoryBean;
@@ -28,10 +28,12 @@ import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
/**
* Bean definition parser for the <gfe:async-event-queue> SDG XML namespace (XSD) element.
* {@link BeanDefinitionParser} for <gfe:async-event-queue> SDG XML Namespace (XSD) Elements.
*
* @author David Turanski
* @author John Blum
* @see org.w3c.dom.Element
* @see org.springframework.beans.factory.support.BeanDefinitionBuilder
* @see org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser
* @see org.springframework.data.gemfire.wan.AsyncEventQueueFactoryBean
*/
@@ -67,6 +69,7 @@ class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
ParsingUtils.setPropertyValue(element, builder, "maximum-queue-memory");
ParsingUtils.setPropertyValue(element, builder, "order-policy");
ParsingUtils.setPropertyValue(element, builder, "parallel");
ParsingUtils.setPropertyValue(element, builder, "pause-event-dispatching");
ParsingUtils.setPropertyValue(element, builder, "persistent");
Element eventFilterElement = DomUtils.getChildElementByTagName(element, "event-filter");
@@ -108,7 +111,6 @@ class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
}
}
/* (non-Javadoc) */
private void parseAsyncEventListener(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
Element asyncEventListenerElement = DomUtils.getChildElementByTagName(element, "async-event-listener");
@@ -123,7 +125,6 @@ class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
}
}
/* (non-Javadoc) */
private void parseCache(Element element, BeanDefinitionBuilder builder) {
String cacheRefAttribute = element.getAttribute("cache-ref");
@@ -132,7 +133,6 @@ class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
builder.addConstructorArgReference(cacheName);
}
/* (non-Javadoc) */
private void parseDiskStore(Element element, BeanDefinitionBuilder builder) {
ParsingUtils.setPropertyValue(element, builder, "disk-store-ref");

42
src/main/java/org/springframework/data/gemfire/util/SpringUtils.java
View File

@@ -14,7 +14,6 @@
* limitations under the License.
*
*/
package org.springframework.data.gemfire.util;
import static org.springframework.data.gemfire.util.ArrayUtils.nullSafeArray;
@@ -23,6 +22,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -35,26 +35,29 @@ import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.util.StringUtils;
/**
* Abstract utility class encapsulating common functionality on {@link Object Objects}
* and other {@link Class Class types}.
* Abstract utility class encapsulating functionality common to {@link Object Objects}, {@link Class Class types}
* and Spring beans.
*
* @author John Blum
* @see java.lang.Class
* @see java.lang.Object
* @see org.springframework.beans.factory.BeanFactory
* @see org.springframework.beans.factory.config.BeanDefinition
* @see org.springframework.beans.factory.config.RuntimeBeanReference
* @since 1.8.0
*/
@SuppressWarnings("unused")
public abstract class SpringUtils {
public static BeanDefinition addDependsOn(BeanDefinition bean, String... beanNames) {
public static BeanDefinition addDependsOn(BeanDefinition beanDefinition, String... beanNames) {
List<String> dependsOnList = new ArrayList<>();
Collections.addAll(dependsOnList, nullSafeArray(bean.getDependsOn(), String.class));
Collections.addAll(dependsOnList, nullSafeArray(beanDefinition.getDependsOn(), String.class));
dependsOnList.addAll(Arrays.asList(nullSafeArray(beanNames, String.class)));
bean.setDependsOn(dependsOnList.toArray(new String[0]));
beanDefinition.setDependsOn(dependsOnList.toArray(new String[0]));
return bean;
return beanDefinition;
}
public static Optional<Object> getPropertyValue(BeanDefinition beanDefinition, String propertyName) {
@@ -102,7 +105,7 @@ public abstract class SpringUtils {
}
public static boolean equalsIgnoreNull(Object obj1, Object obj2) {
return obj1 == null ? obj2 == null : obj1.equals(obj2);
return Objects.equals(obj1, obj2);
}
public static boolean nullOrEquals(Object obj1, Object obj2) {
@@ -121,6 +124,17 @@ public abstract class SpringUtils {
return type != null ? type.getSimpleName() : null;
}
public static boolean safeDoOperation(VoidReturningThrowableOperation operation) {
try {
operation.run();
return true;
}
catch (Throwable cause) {
return false;
}
}
public static <T> T safeGetValue(Supplier<T> valueSupplier) {
return safeGetValue(valueSupplier, (T) null);
}
@@ -143,11 +157,11 @@ public abstract class SpringUtils {
}
}
public static void safeRunOperation(VoidReturningExceptionThrowingOperation operation) {
public static void safeRunOperation(VoidReturningThrowableOperation operation) {
safeRunOperation(operation, cause -> new InvalidDataAccessApiUsageException("Failed to run operation", cause));
}
public static void safeRunOperation(VoidReturningExceptionThrowingOperation operation,
public static void safeRunOperation(VoidReturningThrowableOperation operation,
Function<Throwable, RuntimeException> exceptionConverter) {
try {
@@ -158,8 +172,14 @@ public abstract class SpringUtils {
}
}
/**
* @deprecated use {@link VoidReturningThrowableOperation}.
*/
@Deprecated
public interface VoidReturningExceptionThrowingOperation extends VoidReturningThrowableOperation { }
@FunctionalInterface
public interface VoidReturningExceptionThrowingOperation {
public interface VoidReturningThrowableOperation {
void run() throws Throwable;
}
}

17
src/main/java/org/springframework/data/gemfire/wan/AbstractWANComponentFactoryBean.java
View File

@@ -29,14 +29,14 @@ import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Abstract base class for WAN Gateway components.
* Abstract base class for WAN Gateway objects.
*
* @author David Turanski
* @author John Blum
* @author Udo Kohlmeyer
* @see org.apache.geode.cache.Cache
* @see org.apache.geode.cache.GemFireCache
* @see org.springframework.beans.factory.DisposableBean
* @see org.springframework.beans.factory.FactoryBean
* @see org.springframework.beans.factory.InitializingBean
* @see org.springframework.data.gemfire.support.AbstractFactoryBeanSupport
*/
@@ -64,6 +64,10 @@ public abstract class AbstractWANComponentFactoryBean<T> extends AbstractFactory
this.beanName = beanName;
}
public Cache getCache() {
return this.cache;
}
public void setCache(Cache cache) {
this.cache = cache;
}
@@ -77,16 +81,13 @@ public abstract class AbstractWANComponentFactoryBean<T> extends AbstractFactory
}
public String getName() {
return StringUtils.hasText(this.name)
? this.name
: this.beanName;
return StringUtils.hasText(this.name) ? this.name : this.beanName;
}
@Override
public final void afterPropertiesSet() throws Exception {
Assert.notNull(this.cache, "Cache must not be null");
Assert.notNull(getCache(), "Cache must not be null");
Assert.notNull(getName(), "Name must not be null");
doInit();
@@ -95,6 +96,6 @@ public abstract class AbstractWANComponentFactoryBean<T> extends AbstractFactory
protected abstract void doInit() throws Exception;
@Override
public void destroy() throws Exception { }
public void destroy() { }
}

82
src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java
View File

@@ -15,14 +15,12 @@
*/
package org.springframework.data.gemfire.wan;
import static org.springframework.data.gemfire.util.CollectionUtils.nullSafeList;
import java.util.List;
import java.util.Optional;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.asyncqueue.AsyncEvent;
import org.apache.geode.cache.asyncqueue.AsyncEventListener;
import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory;
@@ -31,6 +29,8 @@ import org.apache.geode.cache.wan.GatewayEventSubstitutionFilter;
import org.apache.geode.cache.wan.GatewaySender;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.data.gemfire.util.CollectionUtils;
import org.springframework.data.gemfire.util.SpringUtils;
import org.springframework.util.Assert;
/**
@@ -56,6 +56,7 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
private Boolean forwardExpirationDestroy;
private Boolean parallel;
private Boolean persistent;
private Boolean pauseEventDispatching;
private Integer batchSize;
private Integer batchTimeInterval;
@@ -106,10 +107,11 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
@Override
protected void doInit() {
Assert.state(this.asyncEventListener != null, "AsyncEventListener must not be null");
AsyncEventListener listener = getAsyncEventListener();
AsyncEventQueueFactory asyncEventQueueFactory =
this.factory != null ? (AsyncEventQueueFactory) this.factory : this.cache.createAsyncEventQueueFactory();
Assert.state(listener != null, "AsyncEventListener must not be null");
AsyncEventQueueFactory asyncEventQueueFactory = resolveAsyncEventQueueFactory();
Optional.ofNullable(this.batchConflationEnabled).ifPresent(asyncEventQueueFactory::setBatchConflationEnabled);
Optional.ofNullable(this.batchSize).ifPresent(asyncEventQueueFactory::setBatchSize);
@@ -122,9 +124,11 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
Optional.ofNullable(this.maximumQueueMemory).ifPresent(asyncEventQueueFactory::setMaximumQueueMemory);
Optional.ofNullable(this.persistent).ifPresent(asyncEventQueueFactory::setPersistent);
asyncEventQueueFactory.setParallel(isParallelEventQueue());
if (isPauseEventDispatching()) {
asyncEventQueueFactory.pauseEventDispatching();
}
nullSafeList(this.gatewayEventFilters).forEach(asyncEventQueueFactory::addGatewayEventFilter);
asyncEventQueueFactory.setParallel(isParallelEventQueue());
if (this.orderPolicy != null) {
@@ -133,21 +137,31 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
asyncEventQueueFactory.setOrderPolicy(this.orderPolicy);
}
setAsyncEventQueue(asyncEventQueueFactory.create(getName(), this.asyncEventListener));
CollectionUtils.nullSafeList(this.gatewayEventFilters).forEach(asyncEventQueueFactory::addGatewayEventFilter);
setAsyncEventQueue(asyncEventQueueFactory.create(getName(), listener));
}
private AsyncEventQueueFactory resolveAsyncEventQueueFactory() {
return this.factory != null ? (AsyncEventQueueFactory) this.factory : this.cache.createAsyncEventQueueFactory();
}
@Override
public void destroy() throws Exception {
public void destroy() {
if (!this.cache.isClosed()) {
try {
this.asyncEventListener.close();
}
catch (CacheClosedException ignore) {
}
if (!getCache().isClosed()) {
SpringUtils.safeDoOperation(() -> this.asyncEventListener.close());
}
}
/**
* Configures the {@link AsyncEventListener} called when {@link AsyncEvent AsyncEvents} are enqueued into
* the {@link AsyncEventQueue} created by this {@link FactoryBean}.
*
* @param listener the configured {@link AsyncEventListener}.
* @throws IllegalStateException if the {@link AsyncEventQueue} has already bean created.
* @see org.apache.geode.cache.asyncqueue.AsyncEventListener
*/
public final void setAsyncEventListener(AsyncEventListener listener) {
Assert.state(this.asyncEventQueue == null,
@@ -156,16 +170,38 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
this.asyncEventListener = listener;
}
/**
* Returns the configured {@link AsyncEventListener} for the {@link AsyncEventQueue}
* returned by this {@link FactoryBean}.
*
* @return the configured {@link AsyncEventListener}.
* @see org.apache.geode.cache.asyncqueue.AsyncEventListener
* @see #setAsyncEventListener(AsyncEventListener)
*/
public AsyncEventListener getAsyncEventListener() {
return this.asyncEventListener;
}
/**
* Configures the {@link AsyncEventQueue} returned by this {@link FactoryBean}.
*
* @param asyncEventQueue overrides {@link AsyncEventQueue} returned by this {@link FactoryBean}.
* @param asyncEventQueue overrides the {@link AsyncEventQueue} returned by this {@link FactoryBean}.
* @see org.apache.geode.cache.asyncqueue.AsyncEventQueue
*/
public void setAsyncEventQueue(AsyncEventQueue asyncEventQueue) {
this.asyncEventQueue = asyncEventQueue;
}
/**
* Returns the {@link AsyncEventQueue} created by this {@link FactoryBean}.
*
* @return a reference to the {@link AsyncEventQueue} created by this {@link FactoryBean}.
* @see org.apache.geode.cache.asyncqueue.AsyncEventQueue
*/
public AsyncEventQueue getAsyncEventQueue() {
return this.asyncEventQueue;
}
/**
* Enable or disable {@link AsyncEventQueue} (AEQ) message conflation.
*
@@ -270,11 +306,19 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
return Boolean.TRUE.equals(parallel);
}
public boolean isSerialEventQueue() {
return !isParallelEventQueue();
public void setPauseEventDispatching(Boolean pauseEventDispatching) {
this.pauseEventDispatching = pauseEventDispatching;
}
public boolean isPauseEventDispatching() {
return Boolean.TRUE.equals(this.pauseEventDispatching);
}
public void setPersistent(Boolean persistent) {
this.persistent = persistent;
}
public boolean isSerialEventQueue() {
return !isParallelEventQueue();
}
}

1
src/main/resources/org/springframework/data/gemfire/config/spring-geode-2.2.xsd
View File

@@ -3028,6 +3028,7 @@ if an inner bean.
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="forward-expiration-destroy" type="xsd:string" default="false" use="optional"/>
<xsd:attribute name="pause-event-dispatching" type="xsd:string" default="false" use="optional"/>
<xsd:attributeGroup ref="commonWANQueueAttributes" />
</xsd:complexType>
<!-- -->