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

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.

Browse Source
This commit is contained in:
John Blum
2013-11-16 14:01:59 -08:00
parent 9a57b0579c
commit f3b22622b1
4 changed files with 195 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

99
src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java
View File

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