From f3b22622b1443bb00db1b95d69807db77f59a96e Mon Sep 17 00:00:00 2001
From: John Blum <jblum@gopivotal.com>
Date: Sat, 16 Nov 2013 14:01:59 -0800
Subject: [PATCH] Fixes JIRA issues SGF-230 and SGF-233 involving the
 persistence on a Async Event Queue or Gateway Sender when a logical Disk
 Store name has been specified, or the disk-synchronous attribute has been
 set, requiring persistence to be enabled when all the user wanted was
 overflow.

---
 .../wan/AsyncEventQueueFactoryBean.java       | 35 +++----
 .../gemfire/wan/GatewaySenderFactoryBean.java | 99 ++++++++++---------
 .../wan/AsyncEventQueueFactoryBeanTest.java   | 64 ++++++++++++
 .../wan/GatewaySenderFactoryBeanTest.java     | 65 ++++++++++++
 4 files changed, 195 insertions(+), 68 deletions(-)

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 8c363be8..c8f4e891 100644
--- a/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java
+++ b/src/main/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBean.java
@@ -15,9 +15,6 @@
  */
 package org.springframework.data.gemfire.wan;
 
-import java.util.Arrays;
-import java.util.List;
-
 import org.springframework.util.Assert;
 
 import com.gemstone.gemfire.cache.Cache;
@@ -49,7 +46,7 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
 	private Integer dispatcherThreads;
 	private Integer maximumQueueMemory;
 
-	private String diskStoreRef;
+	private String diskStoreReference;
 	private String orderPolicy;
 
 	/**
@@ -90,22 +87,6 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
 		AsyncEventQueueFactory asyncEventQueueFactory = (this.factory != null ? (AsyncEventQueueFactory) factory
 			: cache.createAsyncEventQueueFactory());
 
-		if (diskStoreRef != null) {
-			persistent = (persistent == null || persistent);
-			Assert.isTrue(persistent, "Specifying a 'disk store' requires the persistent property to be true.");
-			asyncEventQueueFactory.setDiskStoreName(diskStoreRef);
-		}
-
-		if (diskSynchronous != null) {
-			persistent = (persistent == null || persistent);
-			Assert.isTrue(persistent, "Specifying 'disk synchronous' requires the persistent property to be true.");
-			asyncEventQueueFactory.setDiskSynchronous(diskSynchronous);
-		}
-
-		if (persistent != null) {
-			asyncEventQueueFactory.setPersistent(persistent);
-		}
-
 		if (batchSize != null) {
 			asyncEventQueueFactory.setBatchSize(batchSize);
 		}
@@ -123,6 +104,14 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
 			asyncEventQueueFactory.setDispatcherThreads(dispatcherThreads);
 		}
 
+		if (diskStoreReference != null) {
+			asyncEventQueueFactory.setDiskStoreName(diskStoreReference);
+		}
+
+		if (diskSynchronous != null) {
+			asyncEventQueueFactory.setDiskSynchronous(diskSynchronous);
+		}
+
 		if (maximumQueueMemory != null) {
 			asyncEventQueueFactory.setMaximumQueueMemory(maximumQueueMemory);
 		}
@@ -138,6 +127,10 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
 			asyncEventQueueFactory.setOrderPolicy(Gateway.OrderPolicy.valueOf(orderPolicy.toUpperCase()));
 		}
 
+		if (persistent != null) {
+			asyncEventQueueFactory.setPersistent(persistent);
+		}
+
 		asyncEventQueue = asyncEventQueueFactory.create(getName(), this.asyncEventListener);
 	}
 
@@ -159,7 +152,7 @@ public class AsyncEventQueueFactoryBean extends AbstractWANComponentFactoryBean<
 	}
 
 	public void setDiskStoreRef(String diskStoreRef) {
-		this.diskStoreRef = diskStoreRef;
+		this.diskStoreReference = diskStoreRef;
 	}
 
 	public void setBatchSize(Integer batchSize) {
diff --git a/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java b/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java
index ca5442e7..3d0e95b3 100644
--- a/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java
+++ b/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java
@@ -15,7 +15,6 @@
  */
 package org.springframework.data.gemfire.wan;
 
-import java.util.Arrays;
 import java.util.List;
 
 import org.springframework.context.SmartLifecycle;
@@ -61,7 +60,7 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
 	private Integer socketBufferSize;
 	private Integer socketReadTimeout;
 
-	private String diskStoreRef;
+	private String diskStoreReference;
 	private String orderPolicy;
 
 	/**
@@ -89,24 +88,47 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
 		GatewaySenderFactory gatewaySenderFactory = (this.factory != null ? (GatewaySenderFactory) factory :
 			cache.createGatewaySenderFactory());
 
-		if (diskStoreRef != null) {
-			persistent = (persistent == null || persistent);
-			Assert.isTrue(persistent, "Specifying a disk store requires the persistent property to be true.");
-			gatewaySenderFactory.setDiskStoreName(diskStoreRef);
+		if (alertThreshold != null) {
+			gatewaySenderFactory.setAlertThreshold(alertThreshold);
+		}
+
+		if (batchSize != null) {
+			gatewaySenderFactory.setBatchSize(batchSize);
+		}
+
+		if (batchTimeInterval != null) {
+			gatewaySenderFactory.setBatchTimeInterval(batchTimeInterval);
+		}
+
+		if (diskStoreReference != null) {
+			gatewaySenderFactory.setDiskStoreName(diskStoreReference);
 		}
 
 		if (diskSynchronous != null) {
-			persistent = (persistent == null || persistent);
-			Assert.isTrue(persistent, "Specifying disk synchronous requires the persistent property to be true.");
 			gatewaySenderFactory.setDiskSynchronous(diskSynchronous);
 		}
 
-		if (persistent != null) {
-			gatewaySenderFactory.setPersistenceEnabled(persistent);
+		if (dispatcherThreads != null) {
+			Assert.isTrue(isSerialGatewaySender(),
+				"The number of Dispatcher Threads cannot be specified with a Parallel Gateway Sender Queue.");
+			gatewaySenderFactory.setDispatcherThreads(dispatcherThreads);
 		}
 
-		parallel = Boolean.TRUE.equals(parallel);
-		gatewaySenderFactory.setParallel(parallel);
+		if (enableBatchConflation != null) {
+			gatewaySenderFactory.setBatchConflationEnabled(enableBatchConflation);
+		}
+
+		if (!CollectionUtils.isEmpty(eventFilters)) {
+			for (GatewayEventFilter eventFilter : eventFilters) {
+				gatewaySenderFactory.addGatewayEventFilter(eventFilter);
+			}
+		}
+
+		gatewaySenderFactory.setManualStart(true);
+
+		if (maximumQueueMemory != null) {
+			gatewaySenderFactory.setMaximumQueueMemory(maximumQueueMemory);
+		}
 
 		if (orderPolicy != null) {
 			Assert.isTrue(isSerialGatewaySender(), "Order Policy cannot be used with a Parallel Gateway Sender Queue.");
@@ -117,10 +139,15 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
 			gatewaySenderFactory.setOrderPolicy(Gateway.OrderPolicy.valueOf(orderPolicy.toUpperCase()));
 		}
 
-		if (!CollectionUtils.isEmpty(eventFilters)) {
-			for (GatewayEventFilter eventFilter : eventFilters) {
-				gatewaySenderFactory.addGatewayEventFilter(eventFilter);
-			}
+		gatewaySenderFactory.setParallel(isParallelGatewaySender());
+		gatewaySenderFactory.setPersistenceEnabled(isPersistent());
+
+		if (socketBufferSize != null) {
+			gatewaySenderFactory.setSocketBufferSize(socketBufferSize);
+		}
+
+		if (socketReadTimeout != null) {
+			gatewaySenderFactory.setSocketReadTimeout(socketReadTimeout);
 		}
 
 		if (!CollectionUtils.isEmpty(transportFilters)) {
@@ -129,36 +156,6 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
 			}
 		}
 
-		if (alertThreshold != null) {
-			gatewaySenderFactory.setAlertThreshold(alertThreshold);
-		}
-		if (enableBatchConflation != null) {
-			gatewaySenderFactory.setBatchConflationEnabled(enableBatchConflation);
-		}
-		if (batchSize != null) {
-			gatewaySenderFactory.setBatchSize(batchSize);
-		}
-		if (batchTimeInterval != null) {
-			gatewaySenderFactory.setBatchTimeInterval(batchTimeInterval);
-		}
-		if (dispatcherThreads != null) {
-			Assert.isTrue(isSerialGatewaySender(),
-				"The number of Dispatcher Threads cannot be specified with a Parallel Gateway Sender Queue.");
-			gatewaySenderFactory.setDispatcherThreads(dispatcherThreads);
-		}
-
-		gatewaySenderFactory.setManualStart(true);
-
-		if (maximumQueueMemory != null) {
-			gatewaySenderFactory.setMaximumQueueMemory(maximumQueueMemory);
-		}
-		if (socketBufferSize != null) {
-			gatewaySenderFactory.setSocketBufferSize(socketBufferSize);
-		}
-		if (socketReadTimeout != null) {
-			gatewaySenderFactory.setSocketReadTimeout(socketReadTimeout);
-		}
-
 		GatewaySenderWrapper wrapper = new GatewaySenderWrapper(gatewaySenderFactory.create(getName(),
 			remoteDistributedSystemId));
         wrapper.setManualStart(manualStart);
@@ -194,7 +191,7 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
 	}
 
 	public void setDiskStoreRef(String diskStoreRef) {
-		this.diskStoreRef = diskStoreRef;
+		this.diskStoreReference = diskStoreRef;
 	}
 
 	public void setDiskSynchronous(Boolean diskSynchronous) {
@@ -233,6 +230,14 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
 		this.persistent = persistent;
 	}
 
+	public boolean isNotPersistent() {
+		return !isPersistent();
+	}
+
+	public boolean isPersistent() {
+		return Boolean.TRUE.equals(this.persistent);
+	}
+
 	public void setSocketBufferSize(Integer socketBufferSize) {
 		this.socketBufferSize = socketBufferSize;
 	}
diff --git a/src/test/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBeanTest.java b/src/test/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBeanTest.java
index f811e58e..c271616d 100644
--- a/src/test/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBeanTest.java
+++ b/src/test/java/org/springframework/data/gemfire/wan/AsyncEventQueueFactoryBeanTest.java
@@ -23,6 +23,7 @@ import static org.junit.Assert.assertSame;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Matchers.notNull;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -45,6 +46,7 @@ import com.gemstone.gemfire.cache.util.Gateway;
  * @see org.springframework.data.gemfire.TestUtils
  * @see org.springframework.data.gemfire.wan.AsyncEventQueueFactoryBean
  * @see com.gemstone.gemfire.cache.Cache
+ * @see com.gemstone.gemfire.cache.asyncqueue.AsyncEventListener
  * @see com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueue
  * @see com.gemstone.gemfire.cache.asyncqueue.AsyncEventQueueFactory
  * @since 1.3.3
@@ -90,6 +92,27 @@ public class AsyncEventQueueFactoryBeanTest {
 		if (dispatcherThreads != null) {
 			verify(mockAsyncEventQueueFactory).setDispatcherThreads(eq(dispatcherThreads));
 		}
+
+		String diskStoreReference = TestUtils.readField("diskStoreReference", factoryBean);
+
+		if (diskStoreReference != null) {
+			verify(mockAsyncEventQueueFactory).setDiskStoreName(eq(diskStoreReference));
+		}
+
+		Boolean diskSynchronous = TestUtils.readField("diskSynchronous", factoryBean);
+
+		if (diskSynchronous != null) {
+			verify(mockAsyncEventQueueFactory).setDiskSynchronous(eq(diskSynchronous));
+		}
+
+		Boolean persistent = TestUtils.readField("persistent", factoryBean);
+
+		if (persistent != null) {
+			verify(mockAsyncEventQueueFactory).setPersistent(eq(persistent));
+		}
+		else {
+			verify(mockAsyncEventQueueFactory, never()).setPersistent(true);
+		}
 	}
 
 	@Test
@@ -262,4 +285,45 @@ public class AsyncEventQueueFactoryBeanTest {
 		assertEquals("789", eventQueue.getId());
 	}
 
+	@Test
+	public void testAsyncEventQueueWithOverflowDiskStoreNoPersistence() throws Exception {
+		AsyncEventQueueFactory mockAsyncEventQueueFactory = createMockAsyncEventQueueFactory("123abc");
+
+		AsyncEventQueueFactoryBean factoryBean = new AsyncEventQueueFactoryBean(
+			createMockCacheWithAsyncEventQueueInfrastructure(mockAsyncEventQueueFactory));
+
+		factoryBean.setName("123abc");
+		factoryBean.setAsyncEventListener(createMockAsyncEventListener());
+		factoryBean.setDiskStoreRef("queueOverflowDiskStore");
+		factoryBean.setPersistent(false);
+		factoryBean.doInit();
+
+		verifyExpectations(mockAsyncEventQueueFactory, factoryBean);
+
+		AsyncEventQueue evenQueue = factoryBean.getObject();
+
+		assertNotNull(evenQueue);
+		assertEquals("123abc", evenQueue.getId());
+	}
+
+	@Test
+	public void testAsyncEventQueueWithDiskSynchronousSetPersistenceUnset() throws Exception {
+		AsyncEventQueueFactory mockAsyncEventQueueFactory = createMockAsyncEventQueueFactory("12345");
+
+		AsyncEventQueueFactoryBean factoryBean = new AsyncEventQueueFactoryBean(
+			createMockCacheWithAsyncEventQueueInfrastructure(mockAsyncEventQueueFactory));
+
+		factoryBean.setName("12345");
+		factoryBean.setAsyncEventListener(createMockAsyncEventListener());
+		factoryBean.setDiskSynchronous(true);
+		factoryBean.doInit();
+
+		verifyExpectations(mockAsyncEventQueueFactory, factoryBean);
+
+		AsyncEventQueue evenQueue = factoryBean.getObject();
+
+		assertNotNull(evenQueue);
+		assertEquals("12345", evenQueue.getId());
+	}
+
 }
diff --git a/src/test/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBeanTest.java b/src/test/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBeanTest.java
index bdfb0feb..d0c1cb29 100644
--- a/src/test/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBeanTest.java
+++ b/src/test/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBeanTest.java
@@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -83,6 +84,27 @@ public class GatewaySenderFactoryBeanTest {
 		if (dispatcherThreads != null) {
 			verify(mockGatewaySenderFactory).setDispatcherThreads(eq(dispatcherThreads));
 		}
+
+		String diskStoreReference = TestUtils.readField("diskStoreReference", factoryBean);
+
+		if (diskStoreReference != null) {
+			verify(mockGatewaySenderFactory).setDiskStoreName(eq(diskStoreReference));
+		}
+
+		Boolean diskSynchronous = TestUtils.readField("diskSynchronous", factoryBean);
+
+		if (diskSynchronous != null) {
+			verify(mockGatewaySenderFactory).setDiskSynchronous(eq(diskSynchronous));
+		}
+
+		Boolean persistent = TestUtils.readField("persistent", factoryBean);
+
+		if (persistent != null) {
+			verify(mockGatewaySenderFactory).setPersistenceEnabled(eq(persistent));
+		}
+		else {
+			verify(mockGatewaySenderFactory, never()).setPersistenceEnabled(true);
+		}
 	}
 
 	@Test
@@ -193,4 +215,47 @@ public class GatewaySenderFactoryBeanTest {
 		assertEquals(42, gatewaySender.getRemoteDSId());
 	}
 
+	@Test
+	public void testGatewaySenderWithOverflowDiskStoreNoPersistence() throws Exception {
+		GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g6", 51);
+
+		GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
+			createMockCacheWithGatewayInfrastructure(mockGatewaySenderFactory));
+
+		factoryBean.setName("g6");
+		factoryBean.setRemoteDistributedSystemId(51);
+		factoryBean.setPersistent(false);
+		factoryBean.setDiskStoreRef("queueOverflowDiskStore");
+		factoryBean.doInit();
+
+		verifyExpectations(factoryBean, mockGatewaySenderFactory);
+
+		GatewaySender gatewaySender = factoryBean.getObject();
+
+		assertNotNull(gatewaySender);
+		assertEquals("g6", gatewaySender.getId());
+		assertEquals(51, gatewaySender.getRemoteDSId());
+	}
+
+	@Test
+	public void testGatewaySenderWithDiskSynchronousSetPersistenceUnset() throws Exception {
+		GatewaySenderFactory mockGatewaySenderFactory = createMockGatewaySenderFactory("g7", 51);
+
+		GatewaySenderFactoryBean factoryBean = new GatewaySenderFactoryBean(
+			createMockCacheWithGatewayInfrastructure(mockGatewaySenderFactory));
+
+		factoryBean.setName("g7");
+		factoryBean.setRemoteDistributedSystemId(51);
+		factoryBean.setDiskSynchronous(true);
+		factoryBean.doInit();
+
+		verifyExpectations(factoryBean, mockGatewaySenderFactory);
+
+		GatewaySender gatewaySender = factoryBean.getObject();
+
+		assertNotNull(gatewaySender);
+		assertEquals("g7", gatewaySender.getId());
+		assertEquals(51, gatewaySender.getRemoteDSId());
+	}
+
 }