diff --git a/gradle.properties b/gradle.properties index d3def152..f5cc9521 100644 --- a/gradle.properties +++ b/gradle.properties @@ -7,7 +7,7 @@ slf4jVersion = 1.6.4 # Common libraries springVersion = 3.2.1.RELEASE springDataCommonsVersion = 1.5.0.RELEASE -gemfireVersion = 7.0 +gemfireVersion = 7.0.1 # Testing junitVersion = 4.8.2 diff --git a/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java b/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java index 216b2bd0..ec0e1ce3 100644 --- a/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java +++ b/src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java @@ -57,6 +57,16 @@ public class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser { ParsingUtils.setPropertyValue(element, builder, "persistent"); ParsingUtils.setPropertyValue(element, builder, "parallel"); + + if (ParsingUtils.GEMFIRE_VERSION.compareTo("7.0.1") >= 0 ) { + ParsingUtils.setPropertyValue(element, builder, "batch-conflation-enabled"); + ParsingUtils.setPropertyValue(element, builder, "disk-synchronous"); + ParsingUtils.setPropertyValue(element, builder, "batch-time-interval"); + ParsingUtils.setPropertyValue(element, builder, "disk-synchronous"); + ParsingUtils.setPropertyValue(element, builder, "dispatcher-threads"); + ParsingUtils.setPropertyValue(element, builder, "order-policy"); + } + ParsingUtils.setPropertyValue(element, builder, NAME_ATTRIBUTE); if (!StringUtils.hasText(element.getAttribute(NAME_ATTRIBUTE))) { diff --git a/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java b/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java index 9935aa80..92c2b5e0 100644 --- a/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java @@ -15,6 +15,9 @@ */ package org.springframework.data.gemfire.wan; +import java.util.Arrays; +import java.util.List; + import org.springframework.util.Assert; import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueue; @@ -22,6 +25,7 @@ import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueueFactory; import com.gemstone.gemfire.cache.Cache; import com.gemstone.gemfire.cache.CacheClosedException; import com.gemstone.gemfire.cache.asyncqueue.AsyncEventListener; +import com.gemstone.gemfire.cache.util.Gateway; /** * FactoryBean for creating GemFire {@link AsyncEventQueue}s. @@ -29,6 +33,7 @@ import com.gemstone.gemfire.cache.asyncqueue.AsyncEventListener; * */ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean { + private static List validOrderPolicyValues = Arrays.asList("KEY", "PARTITION", "THREAD"); public void setDiskStoreRef(String diskStoreRef) { this.diskStoreRef = diskStoreRef; @@ -40,14 +45,24 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean< private Integer batchSize; + private Integer dispatcherThreads; + + private Integer batchTimeInterval; + private Integer maximumQueueMemory; private Boolean persistent; private String diskStoreRef; - + private Boolean parallel; + private Boolean diskSynchronous; + + private String orderPolicy; + + private Boolean batchConflationEnabled; + /** * * @param cache the gemfire cache @@ -71,29 +86,58 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean< @Override protected void doInit() { Assert.notNull(asyncEventListener, "AsyncEventListener cannot be null"); + AsyncEventQueueFactory asyncEventQueueFactory = null; if (this.factory == null) { asyncEventQueueFactory = cache.createAsyncEventQueueFactory(); } else { - asyncEventQueueFactory = (AsyncEventQueueFactory)factory; - } - - if (persistent != null) { - asyncEventQueueFactory.setPersistent(persistent); - } - if (batchSize != null) { - asyncEventQueueFactory.setBatchSize(batchSize); + asyncEventQueueFactory = (AsyncEventQueueFactory) factory; } + 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 (diskSynchronous != null) { + persistent = (persistent == null) ? Boolean.TRUE : persistent; + Assert.isTrue(persistent, "specifying a disk synchronous requires persistent property to be true"); + asyncEventQueueFactory.setDiskSynchronous(diskSynchronous); + } + if (persistent != null) { + asyncEventQueueFactory.setPersistent(persistent); + } + + if (batchSize != null) { + asyncEventQueueFactory.setBatchSize(batchSize); + } + + if (batchTimeInterval != null) { + asyncEventQueueFactory.setBatchTimeInterval(batchTimeInterval); + } + + if (batchConflationEnabled != null) { + asyncEventQueueFactory.setBatchConflationEnabled(batchConflationEnabled); + } + + if (dispatcherThreads != null) { + asyncEventQueueFactory.setDispatcherThreads(dispatcherThreads); + } + if (maximumQueueMemory != null) { asyncEventQueueFactory.setMaximumQueueMemory(maximumQueueMemory); } - if( parallel != null ){ - asyncEventQueueFactory.setParallel( parallel ); + + if (parallel != null) { + asyncEventQueueFactory.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"); + asyncEventQueueFactory.setOrderPolicy(Gateway.OrderPolicy.valueOf(orderPolicy.toUpperCase())); } asyncEventQueue = asyncEventQueueFactory.create(getName(), asyncEventListener); @@ -104,8 +148,7 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean< if (!cache.isClosed()) { try { asyncEventListener.close(); - } - catch (CacheClosedException cce) { + } catch (CacheClosedException cce) { // nothing to see folks, move on. } } @@ -122,8 +165,58 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean< public void setPersistent(Boolean persistent) { this.persistent = persistent; } - - public void setParallel(Boolean parallel){ + + public void setParallel(Boolean parallel) { this.parallel = parallel; } + + /** + * @param validOrderPolicyValues the validOrderPolicyValues to set + */ + public static void setValidOrderPolicyValues(List validOrderPolicyValues) { + AsyncEventQueueFactoryBean.validOrderPolicyValues = validOrderPolicyValues; + } + + /** + * @param asyncEventQueue the asyncEventQueue to set + */ + public void setAsyncEventQueue(AsyncEventQueue asyncEventQueue) { + this.asyncEventQueue = asyncEventQueue; + } + + /** + * @param dispatcherThreads the dispatcherThreads to set + */ + public void setDispatcherThreads(Integer dispatcherThreads) { + this.dispatcherThreads = dispatcherThreads; + } + + /** + * @param batchTimeInterval + */ + public void setBatchTimeInterval(Integer batchTimeInterval) { + this.batchTimeInterval = batchTimeInterval; + } + + /** + * + * @param batchConflationEnabled + */ + public void setBatchConflationEnabled(Boolean batchConflationEnabled) { + this.batchConflationEnabled = batchConflationEnabled; + } + + /** + * @param diskSynchronous + */ + public void setDiskSynchronous(Boolean diskSynchronous) { + this.diskSynchronous = diskSynchronous; + } + + /** + * @param orderPolicy + */ + public void setOrderPolicy(String orderPolicy) { + this.orderPolicy = orderPolicy; + } } diff --git a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.3.xsd b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.3.xsd index 9b3ca9dc..7e0a18d1 100755 --- a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.3.xsd +++ b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.3.xsd @@ -2419,45 +2419,45 @@ THREAD:Indicates that events will be parallelized based on the event's originati - - - + + - - - - - + + + + + - - - - - + + + + + - - - - - + + + + + - - - - + ]]> + + + + - + @@ -2593,6 +2593,48 @@ Optionally specifies the GemFire async event queue id. By default this value is ]]> + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java b/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java index ab46efe2..84f53f1e 100644 --- a/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java +++ b/src/test/java/org/springframework/data/gemfire/config/GemfireV7GatewayNamespaceTest.java @@ -89,9 +89,6 @@ public class GemfireV7GatewayNamespaceTest extends RecreatingContextTest { } } - /** - * - */ @Test public void testAsyncEventQueue() { AsyncEventQueue aseq = ctx.getBean("async-event-queue", AsyncEventQueue.class); @@ -100,8 +97,12 @@ public class GemfireV7GatewayNamespaceTest extends RecreatingContextTest { assertTrue(aseq.isParallel()); assertEquals("diskstore", aseq.getDiskStoreName()); assertEquals(50, aseq.getMaximumQueueMemory()); + assertEquals(3, aseq.getBatchTimeInterval()); + assertEquals(OrderPolicy.KEY, aseq.getOrderPolicy()); + assertEquals(true, aseq.isDiskSynchronous()); + assertEquals(true, aseq.isBatchConflationEnabled()); } - + @Test public void testGatewaySender() throws Exception { GatewaySenderFactoryBean gwsfb = ctx.getBean("&gateway-sender", GatewaySenderFactoryBean.class); diff --git a/src/test/java/org/springframework/data/gemfire/test/StubAsyncEventQueueFactory.java b/src/test/java/org/springframework/data/gemfire/test/StubAsyncEventQueueFactory.java index 20d13788..d30f9975 100644 --- a/src/test/java/org/springframework/data/gemfire/test/StubAsyncEventQueueFactory.java +++ b/src/test/java/org/springframework/data/gemfire/test/StubAsyncEventQueueFactory.java @@ -18,6 +18,7 @@ import static org.mockito.Mockito.when; import com.gemstone.gemfire.cache.asyncqueue.AsyncEventListener; import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueue; import com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueueFactory; +import com.gemstone.gemfire.cache.util.Gateway.OrderPolicy; /** * @author David Turanski @@ -82,4 +83,30 @@ public class StubAsyncEventQueueFactory implements AsyncEventQueueFactory { this.parallel = parallel; return this; } + + //The following added in 7.0.1 + public AsyncEventQueueFactory setBatchConflationEnabled(boolean arg0) { + // TODO Auto-generated method stub + return null; + } + + public AsyncEventQueueFactory setBatchTimeInterval(int arg0) { + // TODO Auto-generated method stub + return null; + } + + public AsyncEventQueueFactory setDiskSynchronous(boolean arg0) { + // TODO Auto-generated method stub + return null; + } + + public AsyncEventQueueFactory setDispatcherThreads(int arg0) { + // TODO Auto-generated method stub + return null; + } + + public AsyncEventQueueFactory setOrderPolicy(OrderPolicy arg0) { + // TODO Auto-generated method stub + return null; + } } diff --git a/src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml b/src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml index d08e0778..51dbaa6e 100644 --- a/src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml +++ b/src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml @@ -36,8 +36,17 @@ - +