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

DATAGEODE-368 - Add API to attach additional AsyncEventQueues and GatewaySenders to peer Regions.

Browse Source
This commit is contained in:
John Blum
2020-09-14 13:39:37 -07:00
parent eccf60c737
commit 8a6981fc12
3 changed files with 324 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files

265
spring-data-geode/src/main/java/org/springframework/data/gemfire/PeerRegionFactoryBean.java
View File

@@ -15,12 +15,12 @@
*/
package org.springframework.data.gemfire;
import static java.util.Arrays.stream;
import static org.springframework.data.gemfire.util.ArrayUtils.nullSafeArray;
import static org.springframework.data.gemfire.util.RuntimeExceptionFactory.newIllegalArgumentException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
@@ -56,11 +56,13 @@ import org.springframework.core.io.Resource;
import org.springframework.data.gemfire.client.ClientRegionFactoryBean;
import org.springframework.data.gemfire.eviction.EvictingRegionFactoryBean;
import org.springframework.data.gemfire.expiration.ExpiringRegionFactoryBean;
import org.springframework.data.gemfire.util.ArrayUtils;
import org.springframework.data.gemfire.util.CollectionUtils;
import org.springframework.data.gemfire.util.RegionUtils;
import org.springframework.data.gemfire.util.SpringUtils;
import org.springframework.lang.NonNull;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
@@ -95,6 +97,7 @@ import org.springframework.util.StringUtils;
* @see org.apache.geode.compression.Compressor
* @see org.springframework.beans.factory.DisposableBean
* @see org.springframework.context.SmartLifecycle
* @see org.springframework.data.gemfire.ConfigurableRegionFactoryBean
* @see org.springframework.data.gemfire.ResolvableRegionFactoryBean
* @see org.springframework.data.gemfire.client.ClientRegionFactoryBean
* @see org.springframework.data.gemfire.eviction.EvictingRegionFactoryBean
@@ -113,8 +116,6 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
private Boolean persistent;
private Boolean statisticsEnabled;
private AsyncEventQueue[] asyncEventQueues;
private CacheListener<K, V>[] cacheListeners;
private CacheLoader<K, V> cacheLoader;
@@ -138,7 +139,10 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
private ExpirationAttributes regionIdleTimeout;
private ExpirationAttributes regionTimeToLive;
private GatewaySender[] gatewaySenders;
private List<AsyncEventQueue> asyncEventQueues = new ArrayList<>();
private List<GatewaySender> gatewaySenders = new ArrayList<>();
private List<String> asyncEventQueueIds = new ArrayList<>();
private List<String> gatewaySenderIds = new ArrayList<>();
private RegionAttributes<K, V> attributes;
@@ -150,9 +154,6 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
private String diskStoreName;
private String[] asyncEventQueueIds;
private String[] gatewaySenderIds;
/**
* Creates a new {@link Region} with the given {@link String name}.
*
@@ -175,33 +176,38 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
Region<K, V> region = newRegion(regionFactory, getParent(), regionName);
return enableAsLockGrantor(region);
}
private Region<K, V> enableAsLockGrantor(Region<K, V> region) {
Optional.ofNullable(region)
.filter(it -> it.getAttributes().isLockGrantor())
.ifPresent(Region::becomeLockGrantor);
region = becomeLockGrantor(region);
return region;
}
private Region<K, V> becomeLockGrantor(Region<K, V> region) {
if (isLockGrantor(region)) {
region.becomeLockGrantor();
}
return region;
}
private boolean isLockGrantor(@Nullable Region<K, V> region) {
return region != null && region.getAttributes() != null && region.getAttributes().isLockGrantor();
}
private Region<K, V> newRegion(RegionFactory<K, V> regionFactory, Region<?, ?> parentRegion, String regionName) {
return Optional.ofNullable(parentRegion)
.map(parent -> {
if (parentRegion != null) {
logInfo("Creating Subregion [%1$s] with parent Region [%2$s]", regionName, parent.getName());
logInfo("Creating Subregion [%1$s] with parent Region [%2$s]", regionName, parentRegion.getName());
return regionFactory.createSubregion(parent, regionName);
})
.orElseGet(() -> {
return regionFactory.createSubregion(parentRegion, regionName);
}
else {
logInfo("Created Region [%s]", regionName);
logInfo("Created Region [%s]", regionName);
return regionFactory.create(regionName);
});
return regionFactory.create(regionName);
}
}
private Cache resolveCache(GemFireCache gemfireCache) {
@@ -269,7 +275,8 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
getConfiguredAsyncEventQueueIds().forEach(regionFactory::addAsyncEventQueueId);
stream(nullSafeArray(this.cacheListeners, CacheListener.class)).forEach(regionFactory::addCacheListener);
Arrays.stream(ArrayUtils.nullSafeArray(this.cacheListeners, CacheListener.class))
.forEach(regionFactory::addCacheListener);
Optional.ofNullable(this.cacheLoader).ifPresent(regionFactory::setCacheLoader);
@@ -329,11 +336,12 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
Set<String> asyncEventQueueIds = new HashSet<>();
Arrays.stream(nullSafeArray(this.asyncEventQueues, AsyncEventQueue.class))
CollectionUtils.nullSafeList(this.asyncEventQueues).stream()
.filter(Objects::nonNull)
.map(AsyncEventQueue::getId)
.collect(Collectors.toCollection(() -> asyncEventQueueIds));
Arrays.stream(nullSafeArray(this.asyncEventQueueIds, String.class))
CollectionUtils.nullSafeList(this.asyncEventQueueIds).stream()
.filter(StringUtils::hasText)
.map(String::trim)
.collect(Collectors.toCollection(() -> asyncEventQueueIds));
@@ -345,11 +353,12 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
Set<String> gatewaySenderIds = new HashSet<>();
Arrays.stream(nullSafeArray(this.gatewaySenders, GatewaySender.class))
CollectionUtils.nullSafeList(this.gatewaySenders).stream()
.filter(Objects::nonNull)
.map(GatewaySender::getId)
.collect(Collectors.toCollection(() -> gatewaySenderIds));
Arrays.stream(nullSafeArray(this.gatewaySenderIds, String.class))
CollectionUtils.nullSafeList(this.gatewaySenderIds).stream()
.filter(StringUtils::hasText)
.map(String::trim)
.collect(Collectors.toCollection(() -> gatewaySenderIds));
@@ -358,8 +367,6 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
}
/*
* (non-Javadoc)
*
* This method is not considered part of the PeerRegionFactoryBean API and is strictly used for testing purposes!
*
* NOTE: Cannot pass RegionAttributes.class as the "targetType" in the second invocation of getFieldValue(..)
@@ -437,7 +444,7 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
regionFactory.setEntryIdleTimeout(regionAttributes.getEntryIdleTimeout());
regionFactory.setEntryTimeToLive(regionAttributes.getEntryTimeToLive());
// NOTE: EvictionAttributes are created by certain RegionShortcuts; need the null check!
// NOTE: EvictionAttributes are created by certain RegionShortcuts; null check needed!
if (isUserSpecifiedEvictionAttributes(regionAttributes)) {
regionFactory.setEvictionAttributes(regionAttributes.getEvictionAttributes());
}
@@ -518,14 +525,12 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
allow &= getDataPolicy().withPersistence()
|| (getAttributes() != null && getAttributes().getEvictionAttributes() != null
&& EvictionAction.OVERFLOW_TO_DISK.equals(attributes.getEvictionAttributes().getAction()));
&& EvictionAction.OVERFLOW_TO_DISK.equals(this.attributes.getEvictionAttributes().getAction()));
return allow;
}
/*
* (non-Javadoc)
*
* This method is not part of the PeerRegionFactoryBean API and is strictly used for testing purposes!
*
* @see org.apache.geode.internal.cache.UserSpecifiedRegionAttributes#hasEvictionAttributes
@@ -544,8 +549,6 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
}
/*
* (non-Javadoc)
*
* This method is not part of the PeerRegionFactoryBean API and is strictly used for testing purposes!
*
* @see org.apache.geode.cache.AttributesFactory#validateAttributes(:RegionAttributes)
@@ -652,7 +655,7 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
}
/**
* Closes and destroys the {@link Region}.
* Closes and destroys this {@link Region}.
*
* @throws Exception if {@code destroy()} fails.
* @see org.springframework.beans.factory.DisposableBean
@@ -675,25 +678,82 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
}
/**
* Configures an array of {@link AsyncEventQueue AsyncEventQueues} for this {@link Region} used to perform
* asynchronous data access operations, e.g. {@literal asynchronous write-behind}.
* Configures an array of {@link AsyncEventQueue AsyncEventQueues} for this {@link Region}, which are used
* to perform asynchronous data access operations, e.g. {@literal asynchronous, write-behind operations}.
*
* @param asyncEventQueues array of {@link AsyncEventQueue AsyncEventQueues} used by this {@link Region}
* to perform asynchronous data access operations.
* This method clears any existing, registered {@link AsyncEventQueue AsyncEventQueues} (AEQ) already associated
* with this {@link Region}. Use {@link #addAsyncEventQueues(AsyncEventQueue[])}
* or {@link #addAsyncEventQueueIds(String[])} to append to the existing AEQs already registered instead.
*
* @param asyncEventQueues array of {@link AsyncEventQueue AsyncEventQueues} registered with and used by
* this {@link Region} to perform asynchronous data access operations.
* @see org.apache.geode.cache.asyncqueue.AsyncEventQueue
* @see #addAsyncEventQueues(AsyncEventQueue[])
* @see #addAsyncEventQueueIds(String[])
* @see #setAsyncEventQueueIds(String[])
*/
public void setAsyncEventQueues(AsyncEventQueue[] asyncEventQueues) {
this.asyncEventQueues = asyncEventQueues;
public void setAsyncEventQueues(@NonNull AsyncEventQueue[] asyncEventQueues) {
this.asyncEventQueues.clear();
addAsyncEventQueues(asyncEventQueues);
}
public void setAsyncEventQueueIds(String[] asyncEventQueueIds) {
this.asyncEventQueueIds = asyncEventQueueIds;
/**
* Configures an array of {@link AsyncEventQueue AsyncEventQueues} (AEQ) for this {@link Region}
* by {@link String AEQ ID}.
*
* This method clears any existing, registered {@link AsyncEventQueue AsyncEventQueues} (AEQ) already associated
* with this {@link Region} by {@literal AEQ ID}. Use {@link #addAsyncEventQueues(AsyncEventQueue[])}
* or {@link #addAsyncEventQueueIds(String[])} to append to the existing AEQs already registered instead.
*
* or {@link #addAsyncEventQueueIds(String[])} to append to the existing AEQs already registered instead.
* @param asyncEventQueueIds array of {@link String Strings} specifying {@link String AEQ IDs} to be registered
* with this {@link Region}.
* @see #addAsyncEventQueues(AsyncEventQueue[])
* @see #setAsyncEventQueues(AsyncEventQueue[])
* @see #setAsyncEventQueueIds(String[])
*/
public void setAsyncEventQueueIds(@NonNull String[] asyncEventQueueIds) {
this.asyncEventQueueIds.clear();
addAsyncEventQueueIds(asyncEventQueueIds);
}
/**
* Registers the array of {@link AsyncEventQueue AsyncEventQueues} (AEQ) with this {@link Region} by appending to
* the already existing, registered AEQs for this {@link Region}.
*
* @param asyncEventQueues array of {@link AsyncEventQueue AsyncEventQueues} to register with this {@link Region}.
* @see #addAsyncEventQueueIds(String[])
* @see #setAsyncEventQueues(AsyncEventQueue[])
* @see #setAsyncEventQueueIds(String[])
*/
public void addAsyncEventQueues(@NonNull AsyncEventQueue[] asyncEventQueues) {
Arrays.stream(ArrayUtils.nullSafeArray(asyncEventQueues, AsyncEventQueue.class))
.filter(Objects::nonNull)
.forEach(this.asyncEventQueues::add);
}
/**
* Registers the array of {@link AsyncEventQueue AsyncEventQueues} (AEQ) with this {@link Region}
* by {@link String ID} by appending to the already existing, registered AEQs for this {@link Region}.
*
* @param asyncEventQueueIds array of {@link AsyncEventQueue AsyncEventQueue} {@link String IDs} to register with
* this {@link Region}.
* @see #addAsyncEventQueues(AsyncEventQueue[])
* @see #setAsyncEventQueues(AsyncEventQueue[])
* @see #setAsyncEventQueueIds(String[])
*/
public void addAsyncEventQueueIds(@NonNull String[] asyncEventQueueIds) {
Arrays.stream(ArrayUtils.nullSafeArray(asyncEventQueueIds, String.class))
.filter(StringUtils::hasText)
.forEach(this.asyncEventQueueIds::add);
}
/**
* Sets the {@link RegionAttributes} used to configure this {@link Region}.
*
* Specifying {@link RegionAttributes} allows maximum control in specifying various {@link Region} settings.
* Specifying {@link RegionAttributes} allows full control in configuring various {@link Region} settings.
* Used only when the {@link Region} is created and not when the {@link Region} is looked up.
*
* @param attributes {@link RegionAttributes} used to configure this {@link Region}.
@@ -710,7 +770,10 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
* @see org.apache.geode.cache.RegionAttributes
*/
public RegionAttributes<K, V> getAttributes() {
return Optional.ofNullable(getRegion()).map(Region::getAttributes).orElse(this.attributes);
return Optional.ofNullable(getRegion())
.map(Region::getAttributes)
.orElse(this.attributes);
}
/**
@@ -854,24 +917,83 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
/**
* Configures the {@link GatewaySender GatewaySenders} used to send data and events from this {@link Region}
* to a corresponding {@link Region} in a remote cluster/site.
* to a matching {@link Region} in a remote cluster.
*
* This method clears all existing, registered {@link GatewaySender GatewaySenders} already associated with
* this {@link Region}. Use {@link #addGatewaySenders(GatewaySender[])}
* or {@link #addGatewaySendersIds(String[])} to append to the existing, registered
* {@link GatewaySender GatewaySenders} for this {@link Region}.
*
* @param gatewaySenders {@link GatewaySender GatewaySenders} used to send data and events from this {@link Region}
* to a corresponding {@link Region} in a remote cluster/site.
* to a matching {@link Region} in a remote cluster.
* @see org.apache.geode.cache.wan.GatewaySender
* @see #addGatewaySenders(GatewaySender[])
* @see #addGatewaySendersIds(String[])
* @see #setGatewaySenderIds(String[])
*/
public void setGatewaySenders(GatewaySender[] gatewaySenders) {
this.gatewaySenders = gatewaySenders;
}
public void setGatewaySenderIds(String[] gatewaySenderIds) {
this.gatewaySenderIds = gatewaySenderIds;
public void setGatewaySenders(@NonNull GatewaySender[] gatewaySenders) {
this.gatewaySenders.clear();
addGatewaySenders(gatewaySenders);
}
/**
* Configures whether to enable this {@link Region} with the ability to store data in {@literal off-heap memory}.
* Configures the {@link GatewaySender GatewaySenders} by {@link String ID} used to send data and events from
* this {@link Region} to a matching {@link Region} in a remote cluster.
*
* @param offHeap {@link Boolean} value indicating whether to enable {@literal off-heap memory}
* This method clears all existing, registered {@link GatewaySender GatewaySenders} already associated with
* this {@link Region}. Use {@link #addGatewaySenders(GatewaySender[])}
* or {@link #addGatewaySendersIds(String[])} to append to the existing, registered
* {@link GatewaySender GatewaySenders} for this {@link Region}.
*
* @param gatewaySenderIds {@link String} array containing {@link GatewaySender} {@link String IDs} to register
* with this {@link Region}.
* @see #addGatewaySenders(GatewaySender[])
* @see #addGatewaySendersIds(String[])
* @see #setGatewaySenders(GatewaySender[])
*/
public void setGatewaySenderIds(@NonNull String[] gatewaySenderIds) {
this.gatewaySenderIds.clear();
addGatewaySendersIds(gatewaySenderIds);
}
/**
* Registers the array of {@link GatewaySender GatewaySenders} with this {@link Region} by appending to the already
* existing, registered {@link GatewaySender GatewaySenders} for this {@link Region}.
*
* @param gatewaySenders array of {@link GatewaySender GatewaySenders} to register with this {@link Region}.
* @see org.apache.geode.cache.wan.GatewaySender
* @see #addGatewaySendersIds(String[])
* @see #setGatewaySenders(GatewaySender[])
* @see #setGatewaySenderIds(String[])
*/
public void addGatewaySenders(@NonNull GatewaySender[] gatewaySenders) {
Arrays.stream(ArrayUtils.nullSafeArray(gatewaySenders, GatewaySender.class))
.filter(Objects::nonNull)
.forEach(this.gatewaySenders::add);
}
/**
* Registers the array of {@link GatewaySender} {@link String IDs} with this {@link Region} by appending to
* the already existing, registered {@link GatewaySender GatewaySenders} for this {@link Region}.
*
* @param gatewaySenderIds array of {@link GatewaySender} {@link String IDs} to register with this {@link Region}.
* @see org.apache.geode.cache.wan.GatewaySender
* @see #addGatewaySenders(GatewaySender[])
* @see #setGatewaySenders(GatewaySender[])
* @see #setGatewaySenderIds(String[])
*/
public void addGatewaySendersIds(@NonNull String[] gatewaySenderIds) {
Arrays.stream(ArrayUtils.nullSafeArray(gatewaySenderIds, String.class))
.filter(StringUtils::hasText)
.forEach(this.gatewaySenderIds::add);
}
/**
* Configures this {@link Region} with the capability to store data in {@literal off-heap memory}.
*
* @param offHeap {@link Boolean} value indicating whether to enable the use of {@literal off-heap memory}
* for this {@link Region}.
* @see org.apache.geode.cache.RegionFactory#setOffHeap(boolean)
*/
@@ -882,20 +1004,21 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
/**
* Returns a {@link Boolean} value indicating whether {@literal off-heap memory} was enabled for this {@link Region}.
*
* {@literal Off-heap memory} will be enabled if this method returns a {@literal non-null} {@link Boolean} value
* evaluating to {@literal true}.
* {@literal Off-heap memory} will be enabled for this {@link Region} if this method returns a {@literal non-null},
* {@link Boolean} value evaluating to {@literal true}.
*
* @return a {@link Boolean} value indicating whether {@literal off-heap memory} is enabled for this {@link Region}.
* @return a {@link Boolean} value indicating whether {@literal off-heap memory} use is enabled for
* this {@link Region}.
*/
public Boolean getOffHeap() {
return this.offHeap;
}
/**
* Returns a boolean value indicating whether {@literal off-heap memory} has been enabled for this {@link Region}.
* Returns a boolean value indicating whether {@literal off-heap memory} use was enabled for this {@link Region}.
*
* @return a {@literal boolean} value indicating whether {@literal off-heap memory} has been enabled
* for this {@link Region}.
* @return a {@literal boolean} value indicating whether {@literal off-heap memory} use was enabled for
* this {@link Region}.
* @see #getOffHeap()
*/
public boolean isOffHeap() {
@@ -1024,9 +1147,9 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
@SuppressWarnings("all")
public void start() {
if (!ObjectUtils.isEmpty(this.gatewaySenders)) {
if (!this.gatewaySenders.isEmpty()) {
synchronized(this.gatewaySenders) {
Arrays.stream(this.gatewaySenders)
this.gatewaySenders.stream()
.filter(Objects::nonNull)
.filter(gatewaySender -> !gatewaySender.isManualStart())
.filter(gatewaySender -> !gatewaySender.isRunning())
@@ -1053,11 +1176,9 @@ public abstract class PeerRegionFactoryBean<K, V> extends ConfigurableRegionFact
@SuppressWarnings("all")
public void stop() {
if (!ObjectUtils.isEmpty(this.gatewaySenders)) {
if (!this.gatewaySenders.isEmpty()) {
synchronized (this.gatewaySenders) {
for (Object gatewaySender : this.gatewaySenders) {
((GatewaySender) gatewaySender).stop();
}
this.gatewaySenders.forEach(GatewaySender::stop);
}
}

243
spring-data-geode/src/test/java/org/springframework/data/gemfire/config/xml/GemfireV7GatewayNamespaceTest.java
View File

@@ -16,18 +16,15 @@
package org.springframework.data.gemfire.config.xml;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import java.io.File;
import java.io.FilenameFilter;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.List;
import org.junit.AfterClass;
import org.junit.Test;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.asyncqueue.AsyncEvent;
import org.apache.geode.cache.asyncqueue.AsyncEventListener;
@@ -39,9 +36,6 @@ import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.cache.wan.GatewaySender.OrderPolicy;
import org.apache.geode.cache.wan.GatewayTransportFilter;
import org.junit.AfterClass;
import org.junit.Test;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.data.gemfire.PeerRegionFactoryBean;
import org.springframework.data.gemfire.RecreatingSpringApplicationContextTest;
@@ -50,208 +44,207 @@ import org.springframework.data.gemfire.test.GemfireTestBeanPostProcessor;
import org.springframework.data.gemfire.wan.GatewaySenderFactoryBean;
/**
* The GemfireV7GatewayNamespaceTest class is a test suite of test cases testing the GemFire 7.0 WAN functionality
* by configuring various Gateway senders and receivers.
* Integration Tests testing and asserting GemFire 7.0 WAN functionality and configuration.
*
* @author David Turanski
* @author John Blum
* @see org.springframework.data.gemfire.RecreatingSpringApplicationContextTest
* @see org.junit.Test
* @see org.apache.geode.cache.asyncqueue.AsyncEventQueue
* @see org.apache.geode.cache.wan.GatewayReceiver
* @see org.apache.geode.cache.wan.GatewaySender
* @see org.springframework.data.gemfire.RecreatingSpringApplicationContextTest
*/
@SuppressWarnings("unused")
public class GemfireV7GatewayNamespaceTest extends RecreatingSpringApplicationContextTest {
@AfterClass
@SuppressWarnings("all")
public static void tearDown() {
for (String name : new File(".").list(new FilenameFilter() {
@Override
public boolean accept(File file, String filename) {
return filename.startsWith("BACKUP");
}
})) {
for (String name : new File(".").list((file, filename) -> filename.startsWith("BACKUP"))) {
new File(name).delete();
}
}
/*
* (non-Javadoc)
* @see org.springframework.data.gemfire.RecreatingSpringApplicationContextTest#configureContext(ConfigurableApplicationContext)
*/
@Override
protected <T extends ConfigurableApplicationContext> T configureContext(final T context) {
protected <T extends ConfigurableApplicationContext> T configureContext(T context) {
context.getBeanFactory().addBeanPostProcessor(new GemfireTestBeanPostProcessor());
return context;
}
/*
* (non-Javadoc)
* @see org.springframework.data.gemfire.RecreatingSpringApplicationContextTest#location()
*/
@Override
protected String location() {
return "/org/springframework/data/gemfire/config/xml/gateway-v7-ns.xml";
}
@Test
public void testAsyncEventQueue() {
public void asyncEventQueueConfigurationIsCorrect() {
AsyncEventQueue asyncEventQueue =
this.applicationContext.getBean("async-event-queue", AsyncEventQueue.class);
assertNotNull(asyncEventQueue);
assertTrue(asyncEventQueue.isBatchConflationEnabled());
assertEquals(10, asyncEventQueue.getBatchSize());
assertEquals(3, asyncEventQueue.getBatchTimeInterval());
assertEquals("diskstore", asyncEventQueue.getDiskStoreName());
assertTrue(asyncEventQueue.isDiskSynchronous());
assertEquals(50, asyncEventQueue.getMaximumQueueMemory());
assertEquals(OrderPolicy.KEY, asyncEventQueue.getOrderPolicy());
assertFalse(asyncEventQueue.isParallel());
assertTrue(asyncEventQueue.isPersistent());
assertThat(asyncEventQueue).isNotNull();
assertThat(asyncEventQueue.isBatchConflationEnabled()).isTrue();
assertThat(asyncEventQueue.getBatchSize()).isEqualTo(10);
assertThat(asyncEventQueue.getBatchTimeInterval()).isEqualTo(3);
assertThat(asyncEventQueue.getDiskStoreName()).isEqualTo("diskstore");
assertThat(asyncEventQueue.isDiskSynchronous()).isTrue();
assertThat(asyncEventQueue.getMaximumQueueMemory()).isEqualTo(50);
assertThat(asyncEventQueue.getOrderPolicy()).isEqualTo(OrderPolicy.KEY);
assertThat(asyncEventQueue.isParallel()).isFalse();
assertThat(asyncEventQueue.isPersistent()).isTrue();
}
@Test
public void testGatewaySender() throws Exception {
public void gatewaySenderFactoryBeanConfigurationIsCorrect() throws Exception {
GatewaySenderFactoryBean gatewaySenderFactoryBean =
this.applicationContext.getBean("&gateway-sender", GatewaySenderFactoryBean.class);
assertNotNull(gatewaySenderFactoryBean);
assertNotNull(TestUtils.readField("cache", gatewaySenderFactoryBean));
assertThat((Object) TestUtils.readField("remoteDistributedSystemId", gatewaySenderFactoryBean)).isEqualTo(2);
assertThat((Object) TestUtils.readField("alertThreshold", gatewaySenderFactoryBean)).isEqualTo(10);
assertTrue(Boolean.TRUE.equals(TestUtils.readField("batchConflationEnabled", gatewaySenderFactoryBean)));
assertThat((Object) TestUtils.readField("batchSize", gatewaySenderFactoryBean)).isEqualTo(11);
assertThat((Object) TestUtils.readField("dispatcherThreads", gatewaySenderFactoryBean)).isEqualTo(12);
assertEquals(false, TestUtils.readField("diskSynchronous", gatewaySenderFactoryBean));
assertEquals(true, TestUtils.readField("manualStart", gatewaySenderFactoryBean));
assertThat(gatewaySenderFactoryBean).isNotNull();
assertThat(gatewaySenderFactoryBean.getCache()).isNotNull();
assertThat(gatewaySenderFactoryBean.getRemoteDistributedSystemId()).isEqualTo(2);
assertThat(gatewaySenderFactoryBean.getAlertThreshold()).isEqualTo(10);
assertThat(gatewaySenderFactoryBean.getBatchConflationEnabled()).isTrue();
assertThat(gatewaySenderFactoryBean.getBatchSize()).isEqualTo(11);
assertThat(gatewaySenderFactoryBean.getDispatcherThreads()).isEqualTo(12);
assertThat(gatewaySenderFactoryBean.getDiskSynchronous()).isFalse();
assertThat(gatewaySenderFactoryBean.isManualStart()).isTrue();
List<GatewayEventFilter> eventFilters = TestUtils.readField("eventFilters", gatewaySenderFactoryBean);
assertNotNull(eventFilters);
assertEquals(2, eventFilters.size());
assertTrue(eventFilters.get(0) instanceof TestEventFilter);
assertThat(eventFilters).isNotNull();
assertThat(eventFilters.size()).isEqualTo(2);
assertThat(eventFilters.get(0)).isInstanceOf(TestEventFilter.class);
List<GatewayTransportFilter> transportFilters = TestUtils.readField("transportFilters", gatewaySenderFactoryBean);
assertNotNull(transportFilters);
assertEquals(2, transportFilters.size());
assertTrue(transportFilters.get(0) instanceof TestTransportFilter);
assertThat(transportFilters).isNotNull();
assertThat(transportFilters.size()).isEqualTo(2);
assertThat(transportFilters.get(0)).isInstanceOf(TestTransportFilter.class);
}
@Test
@SuppressWarnings("rawtypes")
public void testInnerGatewaySender() throws Exception {
public void nestedGatewaySenderConfigurationIsCorrect() throws Exception {
Region<?, ?> region = this.applicationContext.getBean("region-inner-gateway-sender", Region.class);
Region<?, ?> region = this.applicationContext.getBean("region-with-nested-gateway-sender", Region.class);
assertNotNull(region.getAttributes().getGatewaySenderIds());
assertEquals(2, region.getAttributes().getGatewaySenderIds().size());
assertThat(region).isNotNull();
assertThat(region.getAttributes()).isNotNull();
assertThat(region.getAttributes().getGatewaySenderIds()).isNotNull();
assertThat(region.getAttributes().getGatewaySenderIds()).hasSize(2);
PeerRegionFactoryBean regionFactoryBean = applicationContext.getBean("&region-inner-gateway-sender", PeerRegionFactoryBean.class);
PeerRegionFactoryBean regionFactoryBean = applicationContext.getBean("&region-with-nested-gateway-sender", PeerRegionFactoryBean.class);
Object[] gatewaySenders = TestUtils.readField("gatewaySenders", regionFactoryBean);
List<GatewaySender> gatewaySenders = TestUtils.readField("gatewaySenders", regionFactoryBean);
assertNotNull(gatewaySenders);
assertEquals(2, gatewaySenders.length);
assertThat(gatewaySenders).isNotNull();
assertThat(gatewaySenders).hasSize(2);
GatewaySender gatewaySender = (GatewaySender) gatewaySenders[0];
GatewaySender gatewaySender = gatewaySenders.get(0);
assertNotNull(gatewaySender);
assertEquals(1, gatewaySender.getRemoteDSId());
assertEquals(false, gatewaySender.isManualStart());
assertEquals(true, gatewaySender.isRunning());
assertEquals(10, gatewaySender.getAlertThreshold());
assertEquals(11, gatewaySender.getBatchSize());
assertEquals(3000, gatewaySender.getBatchTimeInterval());
assertEquals(2, gatewaySender.getDispatcherThreads());
assertEquals("diskstore", gatewaySender.getDiskStoreName());
assertEquals(true, gatewaySender.isDiskSynchronous());
assertTrue(gatewaySender.isBatchConflationEnabled());
assertEquals(50, gatewaySender.getMaximumQueueMemory());
assertEquals(OrderPolicy.THREAD, gatewaySender.getOrderPolicy());
assertTrue(gatewaySender.isPersistenceEnabled());
assertFalse(gatewaySender.isParallel());
assertEquals(16536, gatewaySender.getSocketBufferSize());
assertEquals(3000, gatewaySender.getSocketReadTimeout());
assertThat(gatewaySender).isNotNull();
assertThat(gatewaySender.getRemoteDSId()).isEqualTo(1);
assertThat(gatewaySender.isManualStart()).isFalse();
assertThat(gatewaySender.isRunning()).isTrue();
assertThat(gatewaySender.getAlertThreshold()).isEqualTo(10);
assertThat(gatewaySender.getBatchSize()).isEqualTo(11);
assertThat(gatewaySender.getBatchTimeInterval()).isEqualTo(3000);
assertThat(gatewaySender.getDispatcherThreads()).isEqualTo(2);
assertThat(gatewaySender.getDiskStoreName()).isEqualTo("diskstore");
assertThat(gatewaySender.isDiskSynchronous()).isEqualTo(true);
assertThat(gatewaySender.isBatchConflationEnabled()).isTrue();
assertThat(gatewaySender.getMaximumQueueMemory()).isEqualTo(50);
assertThat(gatewaySender.getOrderPolicy()).isEqualTo(OrderPolicy.THREAD);
assertThat(gatewaySender.isPersistenceEnabled()).isTrue();
assertThat(gatewaySender.isParallel()).isFalse();
assertThat(gatewaySender.getSocketBufferSize()).isEqualTo(16536);
assertThat(gatewaySender.getSocketReadTimeout()).isEqualTo(3000);
List<GatewayEventFilter> eventFilters = gatewaySender.getGatewayEventFilters();
assertNotNull(eventFilters);
assertEquals(1, eventFilters.size());
assertTrue(eventFilters.get(0) instanceof TestEventFilter);
assertThat(eventFilters).isNotNull();
assertThat(eventFilters).hasSize(1);
assertThat(eventFilters.get(0)).isInstanceOf(TestEventFilter.class);
List<GatewayTransportFilter> transportFilters = gatewaySender.getGatewayTransportFilters();
assertNotNull(transportFilters);
assertEquals(1, transportFilters.size());
assertTrue(transportFilters.get(0) instanceof TestTransportFilter);
assertThat(transportFilters).isNotNull();
assertThat(transportFilters).hasSize(1);
assertThat(transportFilters.get(0)).isInstanceOf(TestTransportFilter.class);
}
@Test
public void testGatewaySenderWithEventTransportFilterRefs() throws Exception {
public void gatewaySenderWithEventTransportFilterRefsConfigurationIsCorrect() throws Exception {
GatewaySenderFactoryBean gatewaySenderFactoryBean =
this.applicationContext.getBean("&gateway-sender-with-event-transport-filter-refs",
GatewaySenderFactoryBean.class);
assertNotNull(gatewaySenderFactoryBean);
assertNotNull(TestUtils.readField("cache", gatewaySenderFactoryBean));
assertThat((Integer) TestUtils.readField("remoteDistributedSystemId", gatewaySenderFactoryBean)).isEqualTo(3);
assertTrue(Boolean.TRUE.equals(TestUtils.readField("batchConflationEnabled", gatewaySenderFactoryBean)));
assertThat((Object) TestUtils.readField("batchSize", gatewaySenderFactoryBean)).isEqualTo(50);
assertThat((Object) TestUtils.readField("dispatcherThreads", gatewaySenderFactoryBean)).isEqualTo(10);
assertEquals(true, TestUtils.readField("manualStart", gatewaySenderFactoryBean));
assertThat(gatewaySenderFactoryBean).isNotNull();
assertThat(gatewaySenderFactoryBean.getCache()).isNotNull();
assertThat(gatewaySenderFactoryBean.getRemoteDistributedSystemId()).isEqualTo(3);
assertThat(gatewaySenderFactoryBean.getBatchConflationEnabled()).isTrue();
assertThat(gatewaySenderFactoryBean.getBatchSize()).isEqualTo(50);
assertThat(gatewaySenderFactoryBean.getDispatcherThreads()).isEqualTo(10);
assertThat(gatewaySenderFactoryBean.isManualStart()).isTrue();
List<GatewayEventFilter> eventFilters = TestUtils.readField("eventFilters", gatewaySenderFactoryBean);
assertNotNull(eventFilters);
assertEquals(1, eventFilters.size());
assertTrue(eventFilters.get(0) instanceof TestEventFilter);
assertSame(applicationContext.getBean("event-filter"), eventFilters.get(0));
assertThat(eventFilters).isNotNull();
assertThat(eventFilters).hasSize(1);
assertThat(eventFilters.get(0)).isInstanceOf(TestEventFilter.class);
assertThat(eventFilters.get(0)).isSameAs(applicationContext.getBean("event-filter"));
List<GatewayTransportFilter> transportFilters = TestUtils.readField("transportFilters", gatewaySenderFactoryBean);
List<GatewayTransportFilter> transportFilters =
TestUtils.readField("transportFilters", gatewaySenderFactoryBean);
assertNotNull(transportFilters);
assertEquals(1, transportFilters.size());
assertTrue(transportFilters.get(0) instanceof TestTransportFilter);
assertSame(applicationContext.getBean("transport-filter"), transportFilters.get(0));
assertThat(transportFilters).isNotNull();
assertThat(transportFilters).hasSize(1);
assertThat(transportFilters.get(0)).isInstanceOf(TestTransportFilter.class);
assertThat(transportFilters.get(0)).isSameAs(applicationContext.getBean("transport-filter"));
}
@Test
public void testGatewayReceiver() {
public void gatewayReceiverConfigurationIsCorrect() {
GatewayReceiver gatewayReceiver = this.applicationContext.getBean("gateway-receiver", GatewayReceiver.class);
GatewayReceiver gatewayReceiver =
this.applicationContext.getBean("gateway-receiver", GatewayReceiver.class);
assertNotNull(gatewayReceiver);
assertEquals("192.168.0.1", gatewayReceiver.getBindAddress());
assertEquals(12345, gatewayReceiver.getStartPort());
assertEquals(23456, gatewayReceiver.getEndPort());
assertEquals(3000, gatewayReceiver.getMaximumTimeBetweenPings());
assertEquals(16536, gatewayReceiver.getSocketBufferSize());
assertThat(gatewayReceiver).isNotNull();
assertThat(gatewayReceiver.getBindAddress()).isEqualTo("192.168.0.1");
assertThat(gatewayReceiver.getStartPort()).isEqualTo(12345);
assertThat(gatewayReceiver.getEndPort()).isEqualTo(23456);
assertThat(gatewayReceiver.getMaximumTimeBetweenPings()).isEqualTo(3000);
assertThat(gatewayReceiver.getSocketBufferSize()).isEqualTo(16536);
}
public static class TestAsyncEventListener implements AsyncEventListener {
@Override
public void close() { }
@Override
public boolean processEvents(List<AsyncEvent> events) {
return false;
}
}
@SuppressWarnings("rawtypes")
public static class TestEventFilter implements GatewayEventFilter {
@Override
public void close() { }
@Override
public void afterAcknowledgement(GatewayQueueEvent arg0) { }
public void afterAcknowledgement(GatewayQueueEvent event) { }
@Override
public boolean beforeEnqueue(GatewayQueueEvent arg0) {
public boolean beforeEnqueue(GatewayQueueEvent event) {
return false;
}
@Override
public boolean beforeTransmit(GatewayQueueEvent arg0) {
public boolean beforeTransmit(GatewayQueueEvent event) {
return false;
}
}
@@ -262,25 +255,13 @@ public class GemfireV7GatewayNamespaceTest extends RecreatingSpringApplicationCo
public void close() { }
@Override
public InputStream getInputStream(InputStream arg0) {
public InputStream getInputStream(InputStream in) {
return null;
}
@Override
public OutputStream getOutputStream(OutputStream arg0) {
public OutputStream getOutputStream(OutputStream out) {
return null;
}
}
@SuppressWarnings("rawtypes")
public static class TestAsyncEventListener implements AsyncEventListener {
@Override
public void close() { }
@Override
public boolean processEvents(List<AsyncEvent> arg0) {
return false;
}
}
}

38
spring-data-geode/src/test/resources/org/springframework/data/gemfire/config/xml/gateway-v7-ns.xml
View File

@@ -16,7 +16,25 @@
<gfe:cache properties-ref="gemfireProperties"/>
<gfe:partitioned-region id="region-inner-gateway-sender" >
<gfe:async-event-queue id="async-event-queue"
batch-conflation-enabled="true"
batch-size="10"
batch-time-interval="3"
dispatcher-threads="4"
disk-store-ref="diskstore"
disk-synchronous="true"
maximum-queue-memory="50"
parallel="false"
persistent="true"
order-policy="KEY">
<gfe:async-event-listener>
<bean class="org.springframework.data.gemfire.config.xml.GemfireV7GatewayNamespaceTest.TestAsyncEventListener"/>
</gfe:async-event-listener>
</gfe:async-event-queue>
<gfe:disk-store id="diskstore"/>
<gfe:partitioned-region id="region-with-nested-gateway-sender" >
<gfe:gateway-sender remote-distributed-system-id="1"
manual-start="false"
alert-threshold="10"
@@ -42,24 +60,6 @@
<gfe:gateway-sender-ref bean="gateway-sender"/>
</gfe:partitioned-region>
<gfe:async-event-queue id="async-event-queue"
batch-conflation-enabled="true"
batch-size="10"
batch-time-interval="3"
dispatcher-threads="4"
disk-store-ref="diskstore"
disk-synchronous="true"
maximum-queue-memory="50"
parallel="false"
persistent="true"
order-policy="KEY">
<gfe:async-event-listener>
<bean class="org.springframework.data.gemfire.config.xml.GemfireV7GatewayNamespaceTest.TestAsyncEventListener"/>
</gfe:async-event-listener>
</gfe:async-event-queue>
<gfe:disk-store id="diskstore"/>
<gfe:gateway-receiver id="gateway-receiver" bind-address="192.168.0.1" start-port="12345" end-port="23456"
maximum-time-between-pings="3000" socket-buffer-size="16536">
<gfe:transport-filter>