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

SGF-906 - Refactor ContinuousQueryListenerContainer to use PoolResolver.

Browse Source
This commit is contained in:
John Blum
2019-12-12 17:34:04 -08:00
parent 75a1512611
commit fb7c13fc2c
10 changed files with 438 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/main/java/org/springframework/data/gemfire/GemfireUtils.java
View File

@@ -13,21 +13,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.gemfire;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.internal.GemFireVersion;
import org.w3c.dom.Element;
import org.springframework.data.gemfire.config.support.GemfireFeature;
import org.springframework.data.gemfire.util.RegionUtils;
import org.springframework.util.ClassUtils;
import org.w3c.dom.Element;
/**
* {@link GemfireUtils} is an abstract utility class encapsulating common functionality for accessing features
* and capabilities of Apache Geode or Pivotal GemFire based on version as well as other configuration meta-data.
* and capabilities of Apache Geode or Pivotal GemFire based on version as well as other configuration metadata.
*
* @author John Blum
* @see org.apache.geode.cache.CacheFactory

2
spring-data-geode/src/main/java/org/springframework/data/gemfire/client/PoolResolver.java → src/main/java/org/springframework/data/gemfire/client/PoolResolver.java
View File

@@ -32,7 +32,7 @@ import org.springframework.util.StringUtils;
* as an SPI for different strategies when resolving a {@link Pool}.
*
* @author John Blum
* @see java.lang.FunctionalInterface
* @see FunctionalInterface
* @see org.apache.geode.cache.Region
* @see org.apache.geode.cache.client.Pool
* @since 2.3.0

0
spring-data-geode/src/main/java/org/springframework/data/gemfire/client/support/PoolManagerPoolResolver.java → src/main/java/org/springframework/data/gemfire/client/support/PoolManagerPoolResolver.java
View File

202
src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2011-2020 the original author or authors.
* Copyright 2011-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -13,15 +13,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.gemfire.listener;
import static java.util.stream.StreamSupport.stream;
import static org.springframework.data.gemfire.util.ArrayUtils.nullSafeArray;
import static org.springframework.data.gemfire.util.CollectionUtils.nullSafeIterable;
import static org.springframework.data.gemfire.util.CollectionUtils.nullSafeList;
import static org.springframework.data.gemfire.util.CollectionUtils.nullSafeSet;
import static org.springframework.data.gemfire.util.RuntimeExceptionFactory.newIllegalArgumentException;
import java.util.Arrays;
import java.util.Collections;
@@ -33,10 +28,10 @@ import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.function.Supplier;
import java.util.stream.StreamSupport;
import org.apache.geode.cache.RegionService;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.query.CqAttributes;
import org.apache.geode.cache.query.CqEvent;
import org.apache.geode.cache.query.CqException;
@@ -59,16 +54,22 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.data.gemfire.GemfireQueryException;
import org.springframework.data.gemfire.GemfireUtils;
import org.springframework.data.gemfire.client.PoolResolver;
import org.springframework.data.gemfire.client.support.DefaultableDelegatingPoolAdapter;
import org.springframework.data.gemfire.client.support.DelegatingPoolAdapter;
import org.springframework.data.gemfire.client.support.PoolManagerPoolResolver;
import org.springframework.data.gemfire.config.annotation.ContinuousQueryListenerContainerConfigurer;
import org.springframework.data.gemfire.config.xml.GemfireConstants;
import org.springframework.data.gemfire.util.ArrayUtils;
import org.springframework.data.gemfire.util.CollectionUtils;
import org.springframework.data.gemfire.util.SpringUtils;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
import org.springframework.util.StringUtils;
/**
* Container providing asynchronous processing/handling for Pivotal GemFire / Apache Geode Continuous Queries (CQ).
* Container providing asynchronous processing/handling for Apache Geode Continuous Queries (CQ).
*
* @author Costin Leau
* @author John Blum
@@ -102,6 +103,9 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
public static final String DEFAULT_THREAD_NAME_PREFIX =
String.format("%s-", ContinuousQueryListenerContainer.class.getSimpleName());
// Default PoolResolver uses Apache Geode's PoolManager
protected static final PoolResolver DEFAULT_POOL_RESOLVER = new PoolManagerPoolResolver();
private boolean autoStartup = true;
private volatile boolean initialized = false;
@@ -124,6 +128,8 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
protected final Logger logger = LoggerFactory.getLogger(getClass());
private PoolResolver poolResolver = DEFAULT_POOL_RESOLVER;
private Queue<CqQuery> continuousQueries = new ConcurrentLinkedQueue<>();
private QueryService queryService;
@@ -144,12 +150,19 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
this.initialized = true;
}
/**
* Applies configuration customizations to this {@link ContinuousQueryListenerContainer} from the registered
* composite {@link ContinuousQueryListenerContainerConfigurer} objects.
*
* @see #getCompositeContinuousQueryListenerContainerConfigurer()
* @see #applyContinuousQueryListenerContainerConfigurers(ContinuousQueryListenerContainerConfigurer...)
*/
private void applyContinuousQueryListenerContainerConfigurers() {
applyContinuousQueryListenerContainerConfigurers(getCompositeContinuousQueryListenerContainerConfigurer());
}
/**
* Applies the array of {@link ContinuousQueryListenerContainerConfigurer} objects to customize the configuration
* Applies an array of {@link ContinuousQueryListenerContainerConfigurer} objects to customize the configuration
* of this {@link ContinuousQueryListenerContainer}.
*
* @param configurers array of {@link ContinuousQueryListenerContainerConfigurer} used to customize
@@ -159,12 +172,14 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
protected void applyContinuousQueryListenerContainerConfigurers(
ContinuousQueryListenerContainerConfigurer... configurers) {
applyContinuousQueryListenerContainerConfigurers(Arrays.asList(
nullSafeArray(configurers, ContinuousQueryListenerContainerConfigurer.class)));
List<ContinuousQueryListenerContainerConfigurer> configurerList =
Arrays.asList(ArrayUtils.nullSafeArray(configurers, ContinuousQueryListenerContainerConfigurer.class));
applyContinuousQueryListenerContainerConfigurers(configurerList);
}
/**
* Applies the {@link Iterable} of {@link ContinuousQueryListenerContainerConfigurer} objects to customize
* Applies an {@link Iterable} of {@link ContinuousQueryListenerContainerConfigurer} objects to customize
* the configuration of this {@link ContinuousQueryListenerContainer}.
*
* @param configurers {@link Iterable} of {@link ContinuousQueryListenerContainerConfigurer} used to customize
@@ -174,10 +189,22 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
protected void applyContinuousQueryListenerContainerConfigurers(
Iterable<ContinuousQueryListenerContainerConfigurer> configurers) {
stream(nullSafeIterable(configurers).spliterator(), false)
StreamSupport.stream(CollectionUtils.nullSafeIterable(configurers).spliterator(), false)
.forEach(configurer -> configurer.configure(getBeanName(), this));
}
/**
* Resolves a {@link Pool} object with the given {@link String name} from the configured {@link PoolResolver}.
*
* @param poolName {@link String name} of the {@link Pool} to resolve.
* @return a resolved {@link Pool} object from the given {@link String name}.
* @see org.apache.geode.cache.client.Pool
* @see #getPoolResolver()
*/
@Nullable Pool resolvePool(String poolName) {
return getPoolResolver().resolve(poolName);
}
/**
* Resolves the name of the {@link Pool} configured to handle the registered Continuous Queries.
*
@@ -191,7 +218,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
.filter(StringUtils::hasText)
.orElseGet(() ->
Optional.ofNullable(getBeanFactory())
.filter(it -> it.containsBean(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME))
.filter(it -> SpringUtils.isMatchingBean(it, GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME, Pool.class))
.map(it -> GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME)
.orElse(GemfireUtils.DEFAULT_POOL_NAME));
}
@@ -205,7 +232,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
*
* However, if the {@link BeanFactory} was not configured, or no bean exists in the Spring container
* with the given {@link String name} or of the {@link Pool} type, then the named {@link Pool} is looked up
* in Pivotal GemFire/Apache Geode's {@link PoolManager}.
* in GemFire/Geode's {@link org.apache.geode.cache.client.PoolManager}.
*
* @param poolName {@link String} containing the name of the {@link Pool} to initialize.
* @return the given {@link Pool} name.
@@ -213,13 +240,12 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
String eagerlyInitializePool(String poolName) {
Supplier<String> poolNameResolver = () -> {
Assert.notNull(PoolManager.find(poolName), String.format("No Pool with name [%s] was found", poolName));
Assert.notNull(resolvePool(poolName), String.format("No Pool with name [%s] was found", poolName));
return poolName;
};
return Optional.ofNullable(getBeanFactory())
.filter(it -> it.containsBean(poolName))
.filter(it -> it.isTypeMatch(poolName, Pool.class))
.filter(it -> SpringUtils.isMatchingBean(it, poolName, Pool.class))
.map(it -> {
try {
it.getBean(poolName, Pool.class);
@@ -245,9 +271,13 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
QueryService queryService = getQueryService();
if (queryService == null || StringUtils.hasText(poolName)) {
setQueryService(DefaultableDelegatingPoolAdapter.from(
DelegatingPoolAdapter.from(PoolManager.find(poolName)))
.preferPool().getQueryService(queryService));
Pool resolvedPool = resolvePool(poolName);
DefaultableDelegatingPoolAdapter poolAdapter =
DefaultableDelegatingPoolAdapter.from(DelegatingPoolAdapter.from(resolvedPool));
setQueryService(poolAdapter.preferPool().getQueryService(queryService));
}
return getQueryService();
@@ -287,10 +317,10 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
/**
* Creates a default {@link TaskExecutor}.
*
* <p>Called if no explicit {@link TaskExecutor} has been configured.
* Called if no explicit {@link TaskExecutor} has been configured.
*
* <p>The default implementation builds a {@link SimpleAsyncTaskExecutor} with the specified bean name
* (or the class name, if no bean name is specified) as the Thread name prefix.</p>
* The default implementation builds a {@link SimpleAsyncTaskExecutor} with the specified bean name
* (or the class name, if no bean name is specified) as the Thread name prefix.
*
* @return an instance of the {@link TaskExecutor} used to process CQ events asynchronously.
* @see org.springframework.core.task.SimpleAsyncTaskExecutor
@@ -306,7 +336,11 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
}
/**
* Initializes all the {@link CqQuery Continuous Queries} defined by the {@link ContinuousQueryDefinition defintions}.
* Initializes all the {@link CqQuery Continuous Queries} defined by
* the {@link ContinuousQueryDefinition Continuous Query Defintions}.
*
* @see #getContinuousQueryDefinitions()
* @see #initContinuousQueries(Set)
*/
private void initContinuousQueries() {
initContinuousQueries(getContinuousQueryDefinitions());
@@ -315,17 +349,13 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
private void initContinuousQueries(Set<ContinuousQueryDefinition> continuousQueryDefinitions) {
// Stop the ContinuousQueryListenerContainer if currently running...
if (isRunning()) {
stop();
}
stop();
// Close any existing continuous queries...
closeQueries();
// Add current continuous queries based on the definitions from the configuration...
for (ContinuousQueryDefinition definition : continuousQueryDefinitions) {
addContinuousQuery(definition);
}
continuousQueryDefinitions.forEach(this::addContinuousQuery);
}
/**
@@ -453,8 +483,11 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
* @see #setContinuousQueryListenerContainerConfigurers(List)
*/
public void setContinuousQueryListenerContainerConfigurers(ContinuousQueryListenerContainerConfigurer... configurers) {
setContinuousQueryListenerContainerConfigurers(Arrays.asList(
nullSafeArray(configurers, ContinuousQueryListenerContainerConfigurer.class)));
List<ContinuousQueryListenerContainerConfigurer> configurerList =
Arrays.asList(ArrayUtils.nullSafeArray(configurers, ContinuousQueryListenerContainerConfigurer.class));
setContinuousQueryListenerContainerConfigurers(configurerList);
}
/**
@@ -466,7 +499,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
* @see org.springframework.data.gemfire.config.annotation.ContinuousQueryListenerContainerConfigurer
*/
public void setContinuousQueryListenerContainerConfigurers(List<ContinuousQueryListenerContainerConfigurer> configurers) {
this.cqListenerContainerConfigurers = Optional.ofNullable(configurers).orElseGet(Collections::emptyList);
this.cqListenerContainerConfigurers = CollectionUtils.nullSafeList(configurers);
}
/**
@@ -545,12 +578,35 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
return this.poolName;
}
/**
* Configures the {@link PoolResolver} to resolve {@link Pool} objects by {@link String name}
* from the Apache Geode cache.
*
* @param poolResolver the configured {@link PoolResolver} used to resolve {@link Pool} objects
* by {@link String name}.
* @see org.springframework.data.gemfire.client.PoolResolver
*/
public void setPoolResolver(PoolResolver poolResolver) {
this.poolResolver = poolResolver;
}
/**
* Returns the configured {@link PoolResolver} used to resolve {@link Pool} object by {@link String name}.
*
* @return the configured {@link PoolResolver}.
* @see org.springframework.data.gemfire.client.PoolResolver
*/
public PoolResolver getPoolResolver() {
return this.poolResolver != null ? this.poolResolver : DEFAULT_POOL_RESOLVER;
}
/**
* Attaches the given query definitions.
*
* @param queries set of queries
*/
public void setQueryListeners(Set<ContinuousQueryDefinition> queries) {
getContinuousQueryDefinitions().clear();
getContinuousQueryDefinitions().addAll(nullSafeSet(queries));
}
@@ -628,8 +684,9 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
CqAttributes attributes = definition.toCqAttributes(this::newCqListener);
CqQuery query = (definition.isNamed() ? newNamedContinuousQuery(definition, attributes)
: newUnnamedContinuousQuery(definition, attributes));
CqQuery query = definition.isNamed()
? newNamedContinuousQuery(definition, attributes)
: newUnnamedContinuousQuery(definition, attributes);
return manage(query);
}
@@ -655,7 +712,9 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
}
private CqQuery manage(CqQuery query) {
getContinuousQueries().add(query);
return query;
}
@@ -667,8 +726,8 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
doStart();
this.running = true;
if (logger.isDebugEnabled()) {
logger.debug("Started ContinuousQueryListenerContainer");
if (this.logger.isDebugEnabled()) {
this.logger.debug("Started ContinuousQueryListenerContainer");
}
}
}
@@ -735,22 +794,22 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
boolean active = this.isActive();
if (!active && logger.isDebugEnabled()) {
logger.debug("A CQ listener exception occurred after container shutdown; ErrorHandler will not be invoked", cause);
if (!active && this.logger.isDebugEnabled()) {
this.logger.debug("A CQ listener exception occurred after container shutdown; ErrorHandler will not be invoked", cause);
}
return active;
})
.ifPresent(errorHandler -> errorHandler.handleError(cause));
if (!getErrorHandler().isPresent() && logger.isWarnEnabled()) {
logger.warn("Execution of CQ listener failed; No ErrorHandler was configured", cause);
if (!getErrorHandler().isPresent() && this.logger.isWarnEnabled()) {
this.logger.warn("Execution of CQ listener failed; No ErrorHandler was configured", cause);
}
}
@Override
@SuppressWarnings("all")
public void stop(Runnable callback) {
stop();
callback.run();
}
@@ -763,21 +822,20 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
this.running = false;
}
if (logger.isDebugEnabled()) {
logger.debug("Stopped ContinuousQueryListenerContainer");
if (this.logger.isDebugEnabled()) {
this.logger.debug("Stopped ContinuousQueryListenerContainer");
}
}
void doStop() {
getContinuousQueries().forEach(query -> {
try {
query.stop();
}
catch (Exception cause) {
if (logger.isWarnEnabled()) {
logger.warn(String.format("Cannot stop query [%1$s]; state is [%2$s]",
if (this.logger.isWarnEnabled()) {
this.logger.warn(String.format("Cannot stop query [%1$s]; state is [%2$s]",
query.getName(), query.getState()), cause);
}
}
@@ -785,26 +843,30 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
}
@Override
public void destroy() throws Exception {
public void destroy() {
stop();
closeQueries();
destroyExecutor();
this.initialized = false;
}
private void closeQueries() {
getContinuousQueries().stream().filter(query -> !query.isClosed()).forEach(query -> {
try {
query.close();
}
catch (Exception cause) {
if (logger.isWarnEnabled()) {
logger.warn(String.format("Cannot close query [%1$s]; state is [%2$s]",
query.getName(), query.getState()), cause);
getContinuousQueries().stream()
.filter(query -> !query.isClosed())
.forEach(query -> {
try {
query.close();
}
}
});
catch (Exception cause) {
if (logger.isWarnEnabled()) {
logger.warn(String.format("Cannot close query [%1$s]; state is [%2$s]",
query.getName(), query.getState()), cause);
}
}
});
getContinuousQueries().clear();
}
@@ -813,17 +875,19 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
Optional.ofNullable(getTaskExecutor())
.filter(it -> this.manageExecutor)
.filter(it -> it instanceof DisposableBean)
.filter(DisposableBean.class::isInstance)
.ifPresent(it -> {
try {
((DisposableBean) it).destroy();
if (logger.isDebugEnabled()) {
logger.debug("Stopped internally-managed TaskExecutor {}", it);
if (this.logger.isDebugEnabled()) {
this.logger.debug("Stopped internally-managed TaskExecutor {}", it);
}
}
catch (Exception ignore) {
logger.warn("Failed to properly destroy the managed TaskExecutor {}", it);
catch (Exception cause) {
this.logger.warn("Failed to properly destroy the managed TaskExecutor {}: {}",
it, cause.getMessage());
}
});
}
@@ -833,8 +897,10 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
private final ContinuousQueryListener listener;
protected EventDispatcherAdapter(ContinuousQueryListener listener) {
this.listener = Optional.ofNullable(listener)
.orElseThrow(() -> newIllegalArgumentException("ContinuousQueryListener is required"));
Assert.notNull(listener, "ContinuousQueryListener is required");
this.listener = listener;
}
protected ContinuousQueryListener getListener() {
@@ -849,7 +915,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
dispatchEvent(getListener(), event);
}
public void close() {
}
public void close() { }
}
}

5
src/main/java/org/springframework/data/gemfire/util/CollectionUtils.java
View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.gemfire.util;
import static java.util.stream.StreamSupport.stream;
@@ -39,7 +38,7 @@ import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* {@link CollectionUtils} is a utility class for working with the Java Collections Framework and classes.
* {@link CollectionUtils} is an abstract utility class for working with the Java Collections Framework and classes.
*
* @author John Blum
* @see java.util.Collection
@@ -322,7 +321,7 @@ public abstract class CollectionUtils extends org.springframework.util.Collectio
}
/**
* Sors the elements of the given {@link List} by their natural, {@link Comparable} ordering.
* Sorts the elements of the given {@link List} by their natural, {@link Comparable} ordering.
*
* @param <T> {@link Comparable} class type of the collection elements.
* @param list {@link List} of elements to sort.

19
src/main/java/org/springframework/data/gemfire/util/SpringUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2012-2020 the original author or authors.
* Copyright 2012-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -49,6 +49,23 @@ import org.springframework.util.StringUtils;
@SuppressWarnings("unused")
public abstract class SpringUtils {
/**
* Determines whether a given bean registered in the {@link BeanFactory Spring container} matches by
* both {@link String name} and {@link Class type}.
*
* @param beanFactory {@link BeanFactory Spring container} in which to resolve the bean.
* @param beanName {@link String name} of the bean.
* @param beanType {@link Class type} of the bean.
* @return a boolean value indicating whether the {@link BeanFactory Spring container} contains a bean
* matching by both {@link String name} as well as {@link Class type}.
* @see BeanFactory
* @see Class
* @see String
*/
public static boolean isMatchingBean(BeanFactory beanFactory, String beanName, Class<?> beanType) {
return beanFactory.containsBean(beanName) && beanFactory.isTypeMatch(beanName, beanType);
}
public static BeanDefinition addDependsOn(BeanDefinition beanDefinition, String... beanNames) {
List<String> dependsOnList = new ArrayList<>();

6
spring-data-geode/src/test/java/org/springframework/data/gemfire/client/PoolResolverUnitTests.java → src/test/java/org/springframework/data/gemfire/client/PoolResolverUnitTests.java
View File

@@ -22,7 +22,7 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.verifyZeroInteractions;
import static org.mockito.Mockito.when;
import org.junit.Before;
@@ -42,7 +42,7 @@ import org.apache.geode.cache.client.Pool;
* @see org.junit.Test
* @see org.mockito.Mock
* @see org.mockito.Mockito
* @see org.mockito.junit.MockitoJUnitRunner
* @see MockitoJUnitRunner
* @see org.apache.geode.cache.Region
* @see org.apache.geode.cache.client.Pool
* @since 2.3.0
@@ -73,7 +73,7 @@ public class PoolResolverUnitTests {
assertThat(this.testPoolResolver.resolve(mockRegion)).isEqualTo(mockPool);
verifyNoInteractions(mockPool);
verifyZeroInteractions(mockPool);
verify(mockRegion, times(1)).getAttributes();
verify(mockRegionAttributes, times(1)).getPoolName();
verify(this.testPoolResolver, times(1)).resolve(eq("TestPool"));

0
spring-data-geode/src/test/java/org/springframework/data/gemfire/client/support/PoolManagerPoolResolverIntegrationTests.java → src/test/java/org/springframework/data/gemfire/client/support/PoolManagerPoolResolverIntegrationTests.java
View File

369
src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTests.java → src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerUnitTests.java
View File

@@ -14,16 +14,14 @@
* limitations under the License.
*
*/
package org.springframework.data.gemfire.listener;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isA;
import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
@@ -39,6 +37,12 @@ import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executor;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.apache.geode.cache.RegionService;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.query.CqAttributes;
@@ -48,41 +52,38 @@ import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.CqState;
import org.apache.geode.cache.query.QueryException;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.internal.cache.PoolManagerImpl;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.data.gemfire.GemfireQueryException;
import org.springframework.data.gemfire.GemfireUtils;
import org.springframework.data.gemfire.client.PoolResolver;
import org.springframework.data.gemfire.config.xml.GemfireConstants;
import org.springframework.util.ErrorHandler;
/**
* Unit tests for {@link ContinuousQueryListenerContainer}.
* Unit Tests for {@link ContinuousQueryListenerContainer}.
*
* @author John Blum
* @see org.junit.Rule
* @see org.junit.Test
* @see org.junit.rules.ExpectedException
* @see org.mockito.Mockito
* @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer
* @since 1.8.0
*/
@RunWith(MockitoJUnitRunner.class)
public class ContinuousQueryListenerContainerTests {
public class ContinuousQueryListenerContainerUnitTests {
@Mock
private BeanFactory mockBeanFactory;
private ContinuousQueryListenerContainer cqListenerContainer;
@Before
public void setup() {
this.cqListenerContainer = spy(new ContinuousQueryListenerContainer());
}
private CqQuery mockCqQuery(String name, String query, CqAttributes attributes, boolean durable) {
CqQuery mockQuery = mock(CqQuery.class);
@@ -94,43 +95,42 @@ public class ContinuousQueryListenerContainerTests {
return mockQuery;
}
@Before
public void setup() {
cqListenerContainer = spy(new ContinuousQueryListenerContainer());
}
@Test
public void afterPropertiesSetIsAutoStart() throws Exception {
public void afterPropertiesSetIsAutoStart() {
Pool mockPool = mock(Pool.class);
PoolResolver mockPoolResolver = mock(PoolResolver.class);
QueryService mockQueryService = mock(QueryService.class);
when(mockBeanFactory.containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME))).thenReturn(true);
when(mockBeanFactory.isTypeMatch(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class))).thenReturn(true);
when(mockBeanFactory.getBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class))).thenReturn(mockPool);
when(mockPool.getName()).thenReturn(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME);
when(this.mockBeanFactory.containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME))).thenReturn(true);
when(this.mockBeanFactory.isTypeMatch(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class))).thenReturn(true);
when(this.mockBeanFactory.getBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class))).thenReturn(mockPool);
when(mockPoolResolver.resolve(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME))).thenReturn(mockPool);
when(mockPool.getQueryService()).thenReturn(mockQueryService);
try {
PoolManagerImpl.getPMI().register(mockPool);
this.cqListenerContainer.setAutoStartup(true);
this.cqListenerContainer.setBeanFactory(this.mockBeanFactory);
this.cqListenerContainer.setPoolResolver(mockPoolResolver);
cqListenerContainer.setAutoStartup(true);
cqListenerContainer.setBeanFactory(mockBeanFactory);
cqListenerContainer.afterPropertiesSet();
try {
this.cqListenerContainer.afterPropertiesSet();
}
finally {
assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue();
assertThat(cqListenerContainer.isActive()).isTrue();
assertThat(cqListenerContainer.isAutoStartup()).isTrue();
assertThat(cqListenerContainer.isRunning()).isFalse();
assertThat(cqListenerContainer.getQueryService()).isEqualTo(mockQueryService);
assertThat(cqListenerContainer.getTaskExecutor()).isInstanceOf(Executor.class);
verify(mockBeanFactory, times(2)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME));
verify(mockBeanFactory, times(1)).isTypeMatch(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class));
verify(mockBeanFactory, times(1)).getBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class));
verify(mockPool, times(2)).getName();
assertThat(this.cqListenerContainer.isActive()).isTrue();
assertThat(this.cqListenerContainer.isAutoStartup()).isTrue();
assertThat(this.cqListenerContainer.isRunning()).isFalse();
assertThat(this.cqListenerContainer.getPoolResolver()).isEqualTo(mockPoolResolver);
assertThat(this.cqListenerContainer.getQueryService()).isEqualTo(mockQueryService);
assertThat(this.cqListenerContainer.getTaskExecutor()).isInstanceOf(Executor.class);
verify(this.mockBeanFactory, times(2)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME));
verify(this.mockBeanFactory, times(2)).isTypeMatch(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class));
verify(this.mockBeanFactory, times(1)).getBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class));
verify(mockPoolResolver, times(1)).resolve(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME));
verify(mockPool, times(1)).getQueryService();
verifyZeroInteractions(mockQueryService);
}
@@ -141,45 +141,56 @@ public class ContinuousQueryListenerContainerTests {
Executor mockExecutor = mock(Executor.class);
PoolManagerImpl poolManagerSpy = spy(PoolManagerImpl.getPMI());
Pool mockPool = mock(Pool.class);
PoolResolver mockPoolResolver = mock(PoolResolver.class);
QueryService mockQueryService = mock(QueryService.class);
when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(true);
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
when(this.mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(true);
when(this.mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
when(mockPoolResolver.resolve(eq("TestPool"))).thenReturn(mockPool);
cqListenerContainer.setAutoStartup(false);
cqListenerContainer.setBeanFactory(mockBeanFactory);
cqListenerContainer.setPoolName("TestPool");
cqListenerContainer.setQueryService(mockQueryService);
cqListenerContainer.setTaskExecutor(mockExecutor);
cqListenerContainer.afterPropertiesSet();
this.cqListenerContainer.setAutoStartup(false);
this.cqListenerContainer.setBeanFactory(this.mockBeanFactory);
this.cqListenerContainer.setPoolName("TestPool");
this.cqListenerContainer.setPoolResolver(mockPoolResolver);
this.cqListenerContainer.setQueryService(mockQueryService);
this.cqListenerContainer.setTaskExecutor(mockExecutor);
assertThat(cqListenerContainer.isActive()).isTrue();
assertThat(cqListenerContainer.isAutoStartup()).isFalse();
assertThat(cqListenerContainer.isRunning()).isFalse();
assertThat(cqListenerContainer.getBeanFactory()).isSameAs(mockBeanFactory);
assertThat(cqListenerContainer.getPoolName()).isEqualTo("TestPool");
assertThat(cqListenerContainer.getQueryService()).isSameAs(mockQueryService);
assertThat(cqListenerContainer.getTaskExecutor()).isSameAs(mockExecutor);
try {
this.cqListenerContainer.afterPropertiesSet();
}
finally {
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class));
verify(poolManagerSpy, never()).find(anyString());
verifyZeroInteractions(mockQueryService);
assertThat(this.cqListenerContainer.isActive()).isTrue();
assertThat(this.cqListenerContainer.isAutoStartup()).isFalse();
assertThat(this.cqListenerContainer.isRunning()).isFalse();
assertThat(this.cqListenerContainer.getBeanFactory()).isSameAs(this.mockBeanFactory);
assertThat(this.cqListenerContainer.getPoolName()).isEqualTo("TestPool");
assertThat(this.cqListenerContainer.getPoolResolver()).isEqualTo(mockPoolResolver);
assertThat(this.cqListenerContainer.getQueryService()).isSameAs(mockQueryService);
assertThat(this.cqListenerContainer.getTaskExecutor()).isSameAs(mockExecutor);
verify(this.mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(this.mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
verify(this.mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class));
verify(mockPoolResolver, times(1)).resolve(eq("TestPool"));
verify(mockPool, times(1)).getQueryService();
verifyZeroInteractions(mockQueryService);
}
}
@Test(expected = IllegalStateException.class)
public void afterPropertiesSetThrowsIllegalStateExceptionWhenQueryServiceIsUninitialized() {
when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(true);
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
when(this.mockBeanFactory.containsBean(eq("TestPoolZero"))).thenReturn(true);
when(this.mockBeanFactory.isTypeMatch(eq("TestPoolZero"), eq(Pool.class))).thenReturn(true);
try {
cqListenerContainer.setBeanFactory(mockBeanFactory);
cqListenerContainer.setPoolName("TestPool");
cqListenerContainer.afterPropertiesSet();
this.cqListenerContainer.setBeanFactory(this.mockBeanFactory);
this.cqListenerContainer.setPoolName("TestPoolZero");
this.cqListenerContainer.afterPropertiesSet();
}
catch (IllegalStateException expected) {
@@ -189,71 +200,146 @@ public class ContinuousQueryListenerContainerTests {
throw expected;
}
finally {
assertThat(cqListenerContainer.isActive()).isFalse();
assertThat(cqListenerContainer.isAutoStartup()).isTrue();
assertThat(cqListenerContainer.isRunning()).isFalse();
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class));
verify(mockBeanFactory, times(1)).containsBean(eq("TestPoolZero"));
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPoolZero"), eq(Pool.class));
verify(mockBeanFactory, times(1)).getBean(eq("TestPoolZero"), eq(Pool.class));
}
}
@Test
public void setAndGetPoolResolver() {
PoolResolver mockPoolResolver = mock(PoolResolver.class);
assertThat(this.cqListenerContainer.getPoolResolver())
.isEqualTo(ContinuousQueryListenerContainer.DEFAULT_POOL_RESOLVER);
this.cqListenerContainer.setPoolResolver(mockPoolResolver);
assertThat(this.cqListenerContainer.getPoolResolver()).isEqualTo(mockPoolResolver);
this.cqListenerContainer.setPoolResolver(null);
assertThat(this.cqListenerContainer.getPoolResolver())
.isEqualTo(ContinuousQueryListenerContainer.DEFAULT_POOL_RESOLVER);
verifyZeroInteractions(mockPoolResolver);
}
@Test
public void resolvePoolCallsPoolResolverResolveReturnsNull() {
PoolResolver mockPoolResolver = mock(PoolResolver.class);
when(mockPoolResolver.resolve(anyString())).thenReturn(null);
this.cqListenerContainer.setPoolResolver(mockPoolResolver);
assertThat(this.cqListenerContainer.resolvePool("TestPool")).isNull();
verify(mockPoolResolver, times(1)).resolve(eq("TestPool"));
}
@Test
public void resolvePoolCallsPoolResolverResolveReturnsPool() {
Pool mockPool = mock(Pool.class);
PoolResolver mockPoolResolver = mock(PoolResolver.class);
when(mockPoolResolver.resolve(eq("TestPool"))).thenReturn(mockPool);
this.cqListenerContainer.setPoolResolver(mockPoolResolver);
assertThat(this.cqListenerContainer.resolvePool("TestPool")).isEqualTo(mockPool);
verify(mockPoolResolver, times(1)).resolve(eq("TestPool"));
verifyZeroInteractions(mockPool);
}
@Test
public void resolvePoolNameReturnsConfiguredPoolName() {
cqListenerContainer.setPoolName("TestPool");
this.cqListenerContainer.setPoolName("TestPool");
assertThat(cqListenerContainer.resolvePoolName()).isEqualTo("TestPool");
assertThat(this.cqListenerContainer.resolvePoolName()).isEqualTo("TestPool");
}
@Test
public void resolvePoolNameReturnsGemFireDefaultPoolName() {
public void resolvePoolNameReturnsApacheGeodeDefaultPoolName() {
cqListenerContainer.setPoolName(null);
this.cqListenerContainer.setPoolName(null);
assertThat(cqListenerContainer.resolvePoolName()).isEqualTo(GemfireUtils.DEFAULT_POOL_NAME);
assertThat(this.cqListenerContainer.resolvePoolName()).isEqualTo(GemfireUtils.DEFAULT_POOL_NAME);
}
@Test
public void resolvePoolNameReturnsGemFireDefaultPoolNameWhenBeanFactoryDoesNotContainNamedPool() {
public void resolvePoolNameReturnsApacheGeodeDefaultPoolNameWhenBeanFactoryDoesNotContainNamedPool() {
when(mockBeanFactory.containsBean(anyString())).thenReturn(false);
when(this.mockBeanFactory.containsBean(anyString())).thenReturn(false);
cqListenerContainer.setBeanFactory(mockBeanFactory);
cqListenerContainer.setPoolName("");
this.cqListenerContainer.setBeanFactory(this.mockBeanFactory);
this.cqListenerContainer.setPoolName("");
assertThat(cqListenerContainer.resolvePoolName()).isEqualTo(GemfireUtils.DEFAULT_POOL_NAME);
assertThat(this.cqListenerContainer.resolvePoolName()).isEqualTo(GemfireUtils.DEFAULT_POOL_NAME);
verify(mockBeanFactory, times(1)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME));
verify(this.mockBeanFactory, times(1))
.containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME));
verify(this.mockBeanFactory, never()).isTypeMatch(anyString(), any(Class.class));
}
@Test
public void resolvePoolNameReturnsApacheGeodeDefaultPoolNameWhenBeanFactoryContainsBeanWithNonMatchingPoolType() {
when(this.mockBeanFactory.containsBean(anyString())).thenReturn(true);
when(this.mockBeanFactory.isTypeMatch(anyString(), eq(Pool.class))).thenReturn(false);
this.cqListenerContainer.setBeanFactory(this.mockBeanFactory);
this.cqListenerContainer.setPoolName("");
assertThat(this.cqListenerContainer.resolvePoolName()).isEqualTo(GemfireUtils.DEFAULT_POOL_NAME);
verify(this.mockBeanFactory, times(1))
.containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME));
verify(this.mockBeanFactory, times(1))
.isTypeMatch(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), any(Class.class));
}
@Test
public void resolvePoolNameReturnsSpringDataGemFireDefaultPoolName() {
when(mockBeanFactory.containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME))).thenReturn(true);
when(this.mockBeanFactory.containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME))).thenReturn(true);
when(this.mockBeanFactory.isTypeMatch(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class))).thenReturn(true);
cqListenerContainer.setBeanFactory(mockBeanFactory);
cqListenerContainer.setPoolName(" ");
this.cqListenerContainer.setBeanFactory(this.mockBeanFactory);
this.cqListenerContainer.setPoolName(" ");
assertThat(cqListenerContainer.resolvePoolName()).isEqualTo(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME);
assertThat(this.cqListenerContainer.resolvePoolName()).isEqualTo(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME);
verify(mockBeanFactory, times(1)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME));
verify(this.mockBeanFactory, times(1))
.containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME));
verify(this.mockBeanFactory, times(1))
.isTypeMatch(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class));
}
@Test
public void eagerlyInitializePoolWithGivenPoolName() {
when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(true);
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
when(this.mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(true);
when(this.mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
cqListenerContainer.setBeanFactory(mockBeanFactory);
this.cqListenerContainer.setBeanFactory(this.mockBeanFactory);
assertThat(cqListenerContainer.eagerlyInitializePool("TestPool")).isEqualTo("TestPool");
assertThat(this.cqListenerContainer.eagerlyInitializePool("TestPool")).isEqualTo("TestPool");
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class));
verify(this.mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(this.mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
verify(this.mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class));
}
@Test
@@ -261,23 +347,26 @@ public class ContinuousQueryListenerContainerTests {
Pool mockPool = mock(Pool.class);
when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(false);
when(mockPool.getName()).thenReturn("TestPool");
PoolResolver mockPoolResolver = mock(PoolResolver.class);
when(this.mockBeanFactory.containsBean(eq("TestPoolOne"))).thenReturn(false);
when(mockPoolResolver.resolve(eq("TestPoolOne"))).thenReturn(mockPool);
try {
PoolManagerImpl.getPMI().register(mockPool);
cqListenerContainer.setBeanFactory(mockBeanFactory);
this.cqListenerContainer.setBeanFactory(this.mockBeanFactory);
this.cqListenerContainer.setPoolResolver(mockPoolResolver);
assertThat(cqListenerContainer.eagerlyInitializePool("TestPool")).isEqualTo("TestPool");
assertThat(this.cqListenerContainer.eagerlyInitializePool("TestPoolOne"))
.isEqualTo("TestPoolOne");
}
finally {
assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue();
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, never()).isTypeMatch(eq("TestPool"), eq(Pool.class));
verify(mockBeanFactory, never()).getBean(eq("TestPool"), eq(Pool.class));
verify(mockPool, atLeast(1)).getName();
verify(this.mockBeanFactory, times(1)).containsBean(eq("TestPoolOne"));
verify(this.mockBeanFactory, never()).isTypeMatch(anyString(), eq(Pool.class));
verify(this.mockBeanFactory, never()).getBean(anyString(), eq(Pool.class));
verify(mockPoolResolver, times(1)).resolve(eq("TestPoolOne"));
verifyZeroInteractions(mockPool);
}
}
@@ -286,49 +375,52 @@ public class ContinuousQueryListenerContainerTests {
Pool mockPool = mock(Pool.class);
when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(true);
when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true);
when(mockBeanFactory.getBean(eq("TestPool"), eq(Pool.class)))
.thenThrow(new NoSuchBeanDefinitionException("TEST"));
when(mockPool.getName()).thenReturn("TestPool");
PoolResolver mockPoolResolver = mock(PoolResolver.class);
cqListenerContainer.setBeanFactory(mockBeanFactory);
when(this.mockBeanFactory.containsBean(eq("TestPoolTwo"))).thenReturn(true);
when(this.mockBeanFactory.isTypeMatch(eq("TestPoolTwo"), eq(Pool.class))).thenReturn(true);
when(this.mockBeanFactory.getBean(eq("TestPoolTwo"), eq(Pool.class)))
.thenThrow(new NoSuchBeanDefinitionException("TEST"));
when(mockPoolResolver.resolve(eq("TestPoolTwo"))).thenReturn(mockPool);
this.cqListenerContainer.setBeanFactory(this.mockBeanFactory);
this.cqListenerContainer.setPoolResolver(mockPoolResolver);
try {
PoolManagerImpl.getPMI().register(mockPool);
assertThat(cqListenerContainer.eagerlyInitializePool("TestPool")).isEqualTo("TestPool");
assertThat(this.cqListenerContainer.eagerlyInitializePool("TestPoolTwo"))
.isEqualTo("TestPoolTwo");
}
finally {
assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue();
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class));
verify(mockPool, atLeast(1)).getName();
verify(this.mockBeanFactory, times(1)).containsBean(eq("TestPoolTwo"));
verify(this.mockBeanFactory, times(1)).isTypeMatch(eq("TestPoolTwo"), eq(Pool.class));
verify(this.mockBeanFactory, times(1)).getBean(eq("TestPoolTwo"), eq(Pool.class));
verify(mockPoolResolver, times(1)).resolve(eq("TestPoolTwo"));
verifyZeroInteractions(mockPool);
}
}
@Test(expected = IllegalArgumentException.class)
public void eagerlyInitializePoolThrowsIllegalArgumentExceptionCausedByNoPoolWithGivenName() {
when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(false);
when(this.mockBeanFactory.containsBean(anyString())).thenReturn(false);
try {
cqListenerContainer.setBeanFactory(mockBeanFactory);
cqListenerContainer.eagerlyInitializePool("TestPool");
this.cqListenerContainer.setBeanFactory(this.mockBeanFactory);
this.cqListenerContainer.eagerlyInitializePool("TestPoolThree");
}
catch (IllegalArgumentException expected) {
assertThat(expected).hasMessage("No Pool with name [TestPool] was found");
assertThat(expected).hasMessage("No Pool with name [TestPoolThree] was found");
assertThat(expected).hasNoCause();
throw expected;
}
finally {
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, times(1)).containsBean(eq("TestPoolThree"));
verify(mockBeanFactory, never()).isTypeMatch(anyString(), eq(Pool.class));
verify(mockBeanFactory, never()).getBean(eq("TestPool"), eq(Pool.class));
verify(mockBeanFactory, never()).getBean(anyString(), eq(Pool.class));
}
}
@@ -354,21 +446,23 @@ public class ContinuousQueryListenerContainerTests {
Pool mockPool = mock(Pool.class);
PoolResolver mockPoolResolver = mock(PoolResolver.class);
QueryService mockQueryService = mock(QueryService.class);
when(mockPool.getName()).thenReturn("TestPool");
when(mockPoolResolver.resolve(eq("TestPoolFour"))).thenReturn(mockPool);
when(mockPool.getQueryService()).thenReturn(mockQueryService);
try {
PoolManagerImpl.getPMI().register(mockPool);
this.cqListenerContainer.setPoolResolver(mockPoolResolver);
assertThat(cqListenerContainer.getQueryService()).isNull();
assertThat(cqListenerContainer.initQueryService("TestPool")).isEqualTo(mockQueryService);
try {
assertThat(this.cqListenerContainer.getQueryService()).isNull();
assertThat(this.cqListenerContainer.initQueryService("TestPoolFour")).isEqualTo(mockQueryService);
}
finally {
assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue();
verify(mockPool, atLeast(1)).getName();
verify(mockPoolResolver, times(1)).resolve(eq("TestPoolFour"));
verify(mockPool, times(1)).getQueryService();
verifyZeroInteractions(mockQueryService);
}
@@ -379,24 +473,27 @@ public class ContinuousQueryListenerContainerTests {
Pool mockPool = mock(Pool.class);
PoolResolver mockPoolResolver = mock(PoolResolver.class);
QueryService mockQueryServiceOne = mock(QueryService.class);
QueryService mockQueryServiceTwo = mock(QueryService.class);
when(mockPool.getName()).thenReturn("TestPool");
when(mockPoolResolver.resolve(eq("TestPoolFive"))).thenReturn(mockPool);
when(mockPool.getQueryService()).thenReturn(mockQueryServiceOne);
this.cqListenerContainer.setPoolResolver(mockPoolResolver);
this.cqListenerContainer.setQueryService(mockQueryServiceTwo);
try {
PoolManagerImpl.getPMI().register(mockPool);
cqListenerContainer.setQueryService(mockQueryServiceTwo);
assertThat(cqListenerContainer.getQueryService()).isSameAs(mockQueryServiceTwo);
assertThat(cqListenerContainer.initQueryService("TestPool")).isEqualTo(mockQueryServiceOne);
assertThat(this.cqListenerContainer.getPoolResolver()).isSameAs(mockPoolResolver);
assertThat(this.cqListenerContainer.getQueryService()).isSameAs(mockQueryServiceTwo);
assertThat(this.cqListenerContainer.initQueryService("TestPoolFive"))
.isEqualTo(mockQueryServiceOne);
}
finally {
assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue();
verify(mockPool, atLeast(1)).getName();
verify(mockPoolResolver, times(1)).resolve(eq("TestPoolFive"));
verify(mockPool, times(1)).getQueryService();
verifyZeroInteractions(mockQueryServiceOne);
verifyZeroInteractions(mockQueryServiceTwo);

45
src/test/java/org/springframework/data/gemfire/util/SpringUtilsUnitTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.data.gemfire.util;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doReturn;
@@ -41,8 +42,11 @@ import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.apache.geode.cache.client.Pool;
import org.springframework.beans.MutablePropertyValues;
import org.springframework.beans.PropertyValue;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.dao.InvalidDataAccessApiUsageException;
@@ -65,6 +69,47 @@ public class SpringUtilsUnitTests {
@Mock
private BeanDefinition mockBeanDefinition;
@Test
public void isMatchingBeanReturnsTrue() {
BeanFactory mockBeanFactory = mock(BeanFactory.class);
when(mockBeanFactory.containsBean(anyString())).thenReturn(true);
when(mockBeanFactory.isTypeMatch(anyString(), any(Class.class))).thenReturn(true);
assertThat(SpringUtils.isMatchingBean(mockBeanFactory, "TestPool", Pool.class)).isTrue();
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
}
@Test
public void isMatchingBeanReturnsFalseWhenBeanFactoryDoesNotContainBeanByName() {
BeanFactory mockBeanFactory = mock(BeanFactory.class);
when(mockBeanFactory.containsBean(anyString())).thenReturn(false);
assertThat(SpringUtils.isMatchingBean(mockBeanFactory, "TestPool", Pool.class)).isFalse();
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, never()).isTypeMatch(anyString(), any(Class.class));
}
@Test
public void isMatchingBeanReturnsFalseWhenBeanIsNotATypeMatch() {
BeanFactory mockBeanFactory = mock(BeanFactory.class);
when(mockBeanFactory.containsBean(anyString())).thenReturn(true);
when(mockBeanFactory.isTypeMatch(anyString(), any(Class.class))).thenReturn(false);
assertThat(SpringUtils.isMatchingBean(mockBeanFactory, "TestPool", Pool.class)).isFalse();
verify(mockBeanFactory, times(1)).containsBean(eq("TestPool"));
verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class));
}
@Test
public void addDependsOnToExistingDependencies() {