SGF-164,SGF-165 upgrade to Gemfire 7.0.1

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

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))) {

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<AsyncEventQueue> {
+	private static List<String> 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<String> 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;
+	}
 }

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
     </xsd:complexType>
     <!-- -->
     <xsd:attributeGroup name="commonWANQueueAttributes">
-    	<xsd:attribute name="batch-size" type="xsd:string"
-    		use="optional">
-    		<xsd:annotation>
-    			<xsd:documentation><![CDATA[
+    		<xsd:attribute name="batch-size" type="xsd:string"
+    			use="optional">
+    			<xsd:annotation>
+    				<xsd:documentation><![CDATA[
 Specifies the batch size
-            ]]></xsd:documentation>
-    		</xsd:annotation>
-    	</xsd:attribute>
-    	<xsd:attribute name="persistent" type="xsd:string"
-    		use="optional">
-    		<xsd:annotation>
-    			<xsd:documentation><![CDATA[
+            	]]></xsd:documentation>
+    			</xsd:annotation>
+    		</xsd:attribute>
+    		<xsd:attribute name="persistent" type="xsd:string"
+    			use="optional">
+    			<xsd:annotation>
+    				<xsd:documentation><![CDATA[
 Specifies whether persistence is enabled: true or false(default)
-            ]]></xsd:documentation>
-    		</xsd:annotation>
-    	</xsd:attribute>
-    	<xsd:attribute name="disk-store-ref" type="xsd:string"
-    		use="optional">
-    		<xsd:annotation>
-    			<xsd:documentation><![CDATA[
+            	]]></xsd:documentation>
+    			</xsd:annotation>
+    		</xsd:attribute>
+    		<xsd:attribute name="disk-store-ref" type="xsd:string"
+    			use="optional">
+    			<xsd:annotation>
+    				<xsd:documentation><![CDATA[
 Indicates the id of disk store to use for persistence
-            ]]></xsd:documentation>
-    		</xsd:annotation>
-    	</xsd:attribute>
-    	<xsd:attribute name="maximum-queue-memory" type="xsd:string"
-    		use="optional">
-    		<xsd:annotation>
-    			<xsd:documentation><![CDATA[
+            	]]></xsd:documentation>
+    			</xsd:annotation>
+    		</xsd:attribute>
+    		<xsd:attribute name="maximum-queue-memory" type="xsd:string"
+    			use="optional">
+    			<xsd:annotation>
+    				<xsd:documentation><![CDATA[
 Specifies the maximum memory in MB to allocate for the queue
-            ]]></xsd:documentation>
-    		</xsd:annotation>
-    	</xsd:attribute>
-    	<xsd:attribute name="parallel" type="xsd:string">
-    	<xsd:annotation>
+            	]]></xsd:documentation>
+ 	   		</xsd:annotation>
+    		</xsd:attribute>
+    		<xsd:attribute name="parallel" type="xsd:string">
+    		<xsd:annotation>
                 <xsd:documentation><![CDATA[
                 A value of "false" or "true" that specifies the type of queue GemFire creates, serial or parallel.
             ]]></xsd:documentation>
             </xsd:annotation>
-    	</xsd:attribute>
+    		</xsd:attribute>
     </xsd:attributeGroup>
     <!-- -->
     <xsd:complexType name="gatewayReceiverType">
@@ -2593,6 +2593,48 @@ Optionally specifies the GemFire async event queue id. By default this value is
                   ]]></xsd:documentation>
             </xsd:annotation>
         </xsd:attribute>
+        <xsd:attribute name="batch-conflation-enabled" type="xsd:string" use="optional">
+             <xsd:annotation>
+                <xsd:documentation><![CDATA[
+Optionally specifies whether to conflate queued events (true or false)
+ (only available in Gemfire 7.0.1 + )
+                  ]]></xsd:documentation>
+            </xsd:annotation>
+        </xsd:attribute>
+        <xsd:attribute name="batch-time-interval" type="xsd:string" use="optional">
+             <xsd:annotation>
+                <xsd:documentation><![CDATA[
+Specifies the maximum time interval that can elapse before a partial batch is sent from a the AsyncEventQueue
+ (only available in Gemfire 7.0.1 + )
+                  ]]></xsd:documentation>
+            </xsd:annotation>
+        </xsd:attribute>
+        <xsd:attribute name="disk-synchronous" type="xsd:string" use="optional">
+             <xsd:annotation>
+                <xsd:documentation><![CDATA[
+Specifies if disk writes should be synchronous (true or false) (only available in Gemfire 7.0.1 + )
+                  ]]></xsd:documentation>
+            </xsd:annotation>
+        </xsd:attribute> 
+        <xsd:attribute name="dispatcher-threads" type="xsd:string" use="optional">
+             <xsd:annotation>
+                <xsd:documentation><![CDATA[
+Specifies the number of dispatcher threads to use (only available in Gemfire 7.0.1 + )
+                  ]]></xsd:documentation>
+            </xsd:annotation>
+        </xsd:attribute>
+       <xsd:attribute name="order-policy" type="xsd:string"
+            use="optional">
+            <xsd:annotation>
+                <xsd:documentation><![CDATA[
+(Only available in GemFire 7.0.1 + )
+Specifies the order policy - This only applies if parallel is enabled:
+KEY: Indicates that events will be parallelized based on the event's key,
+PARTITION:Indicates that events will be parallelized based on the event's: partition (using the PartitionResolver)
+THREAD:Indicates that events will be parallelized based on the event's originating member and thread
+            ]]></xsd:documentation>
+            </xsd:annotation>
+        </xsd:attribute>
         <xsd:attributeGroup ref="commonWANQueueAttributes" />
     </xsd:complexType>
     <!-- -->

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

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

src/test/resources/org/springframework/data/gemfire/config/gateway-v7-ns.xml

@@ -36,8 +36,17 @@
         <gfe:gateway-sender-ref bean="gateway-sender"/>
     </gfe:partitioned-region>
     
-    <gfe:async-event-queue id="async-event-queue" batch-size="10" persistent="true" disk-store-ref="diskstore" 
-    maximum-queue-memory="50" parallel="true">
+    <gfe:async-event-queue id="async-event-queue" 
+    batch-size="10" 
+    persistent="true" 
+    disk-store-ref="diskstore" 
+    maximum-queue-memory="50" 
+    parallel="true"
+    batch-conflation-enabled="true"
+    batch-time-interval="3"
+    dispatcher-threads="4"
+    disk-synchronous="true"
+    order-policy="KEY">
         <gfe:async-event-listener>
           <bean class="org.springframework.data.gemfire.config.GemfireV7GatewayNamespaceTest.TestAsyncEventListener"/>
         </gfe:async-event-listener>