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

Introduce more reliable coordination between a GemFire client/server during integration tests.

Browse Source 
Fixes gh-672
This commit is contained in:
John Blum
2017-01-18 13:58:55 -08:00
parent 5c6565bd9c
commit b3706addbb
11 changed files with 329 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
samples/javaconfig/gemfire-clientserver/src/main/java/sample/ClientConfig.java
View File

@@ -20,8 +20,14 @@ import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.client.ClientCache;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.internal.PoolImpl;
import org.apache.geode.management.membership.ClientMembership;
import org.apache.geode.management.membership.ClientMembershipEvent;
import org.apache.geode.management.membership.ClientMembershipListenerAdapter;
@@ -32,8 +38,10 @@ import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.annotation.Bean;
import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
import org.springframework.data.gemfire.client.ClientCacheFactoryBean;
import org.springframework.data.gemfire.config.xml.GemfireConstants;
import org.springframework.data.gemfire.support.ConnectionEndpoint;
import org.springframework.session.data.gemfire.config.annotation.web.http.EnableGemFireHttpSession;
import org.springframework.session.data.gemfire.config.annotation.web.http.GemFireHttpSessionConfiguration;
import org.springframework.util.Assert;
// tag::class[]
@@ -44,7 +52,8 @@ public class ClientConfig {
static final CountDownLatch LATCH = new CountDownLatch(1);
static final String DEFAULT_GEMFIRE_LOG_LEVEL = "config";
static final String DEFAULT_GEMFIRE_LOG_LEVEL = "warning";
static final String GEMFIRE_DEFAULT_POOL_NAME = "DEFAULT";
static final String PROXY_HOST = "dummy.example.com";
static final String PROXY_PORT = "3128";
@@ -62,11 +71,11 @@ public class ClientConfig {
}
String applicationName() {
return "samples:httpsession-gemfire-clientserver:".concat(getClass().getSimpleName());
return "spring-session-data-gemfire-clientserver-javaconfig-sample:".concat(getClass().getSimpleName());
}
String logLevel() {
return System.getProperty("sample.httpsession.gemfire.log-level", DEFAULT_GEMFIRE_LOG_LEVEL);
return System.getProperty("spring.session.data.gemfire.log-level", DEFAULT_GEMFIRE_LOG_LEVEL);
}
@Bean
@@ -112,19 +121,20 @@ public class ClientConfig {
}
@Bean
BeanPostProcessor gemfireCacheServerReadyBeanPostProcessor(
@Value("${spring.session.data.gemfire.host:" + ServerConfig.SERVER_HOST + "}") final String host,
@Value("${spring.session.data.gemfire.port:" + ServerConfig.SERVER_PORT + "}") final int port) { // <5>
BeanPostProcessor gemfireClientServerReadyBeanPostProcessor(
@Value("${spring-session-data-gemfire.cache.server.host:localhost}") final String host,
@Value("${spring-session-data-gemfire.cache.server.port:12480}") final int port) { // <5>
return new BeanPostProcessor() {
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof Region) {
try {
boolean didNotTimeout = LATCH.await(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
private final AtomicBoolean checkGemFireServerIsRunning = new AtomicBoolean(true);
private final AtomicReference<Pool> defaultPool = new AtomicReference<Pool>(null);
Assert.state(didNotTimeout, String.format(
"GemFire Cache Server failed to start on host [%s] and port [%d]", host, port));
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
if (shouldCheckWhetherGemFireServerIsRunning(bean, beanName)) {
try {
validateCacheClientNotified();
validateCacheClientSubscriptionQueueConnectionEstablished();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -134,6 +144,69 @@ public class ClientConfig {
return bean;
}
private boolean shouldCheckWhetherGemFireServerIsRunning(Object bean, String beanName) {
return (isGemFireRegion(bean, beanName)
? checkGemFireServerIsRunning.compareAndSet(true, false)
: whenGemFireCache(bean, beanName));
}
private boolean isGemFireRegion(Object bean, String beanName) {
return (GemFireHttpSessionConfiguration.DEFAULT_SPRING_SESSION_GEMFIRE_REGION_NAME.equals(beanName)
|| bean instanceof Region);
}
private boolean whenGemFireCache(Object bean, String beanName) {
if (bean instanceof ClientCache) {
defaultPool.compareAndSet(null, ((ClientCache) bean).getDefaultPool());
}
return false;
}
private void validateCacheClientNotified() throws InterruptedException {
boolean didNotTimeout = LATCH.await(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
Assert.state(didNotTimeout, String.format(
"GemFire Cache Server failed to start on host [%s] and port [%d]", host, port));
}
@SuppressWarnings("all")
private void validateCacheClientSubscriptionQueueConnectionEstablished() throws InterruptedException {
boolean cacheClientSubscriptionQueueConnectionEstablished = false;
Pool pool = defaultIfNull(this.defaultPool.get(), GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME,
GEMFIRE_DEFAULT_POOL_NAME);
if (pool instanceof PoolImpl) {
long timeout = (System.currentTimeMillis() + DEFAULT_TIMEOUT);
while (System.currentTimeMillis() < timeout
&& !((PoolImpl) pool).isPrimaryUpdaterAlive()) {
synchronized (pool) {
TimeUnit.MILLISECONDS.timedWait(pool, 500L);
}
}
cacheClientSubscriptionQueueConnectionEstablished |=
((PoolImpl) pool).isPrimaryUpdaterAlive();
}
Assert.state(cacheClientSubscriptionQueueConnectionEstablished, String.format(
"Cache client subscription queue connection not established; GemFire Pool was [%s];"
+ " GemFire Pool configuration was [locators = %s, servers = %s]",
pool, pool.getLocators(), pool.getServers()));
}
private Pool defaultIfNull(Pool pool, String... poolNames) {
for (String poolName : poolNames) {
pool = (pool != null ? pool : PoolManager.find(poolName));
}
return pool;
}
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
return bean;
}