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

SGF-693 - Allow the QueryService to be configured as a Spring bean in the ContinuousQuery Annotation support.

Browse Source
This commit is contained in:
John Blum
2017-11-21 23:00:22 -08:00
parent 5ab358bd0a
commit caa6ff79fd
4 changed files with 127 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
src/main/java/org/springframework/data/gemfire/config/annotation/ContinuousQueryConfiguration.java
View File

@@ -29,6 +29,8 @@ import java.util.concurrent.Executor;
import java.util.stream.Collectors;
import org.apache.geode.cache.GemFireCache;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.QueryService;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
@@ -45,17 +47,19 @@ import org.springframework.data.gemfire.listener.ContinuousQueryDefinition;
import org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer;
import org.springframework.data.gemfire.listener.annotation.ContinuousQuery;
import org.springframework.data.gemfire.util.CacheUtils;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
import org.springframework.util.StringUtils;
/**
* The {@link ContinuousQueryConfiguration} class is a Spring {@link Configuration @Configuration} class enabling
* Continuous Query (CQ) GemFire/Geode capabilities in this cache client application.
* Continuous Query (CQ) Pivotal GemFire/Apache Geode capabilities in this cache client application.
*
* @author John Blum
* @see java.util.concurrent.Executor
* @see org.apache.geode.cache.GemFireCache
* @see org.apache.geode.cache.query.CqQuery
* @see org.apache.geode.cache.query.QueryService
* @see org.springframework.beans.factory.config.BeanPostProcessor
* @see org.springframework.context.annotation.Bean
* @see org.springframework.context.annotation.Configuration
@@ -67,6 +71,7 @@ import org.springframework.util.StringUtils;
* @see org.springframework.data.gemfire.listener.ContinuousQueryListener
* @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer
* @see org.springframework.data.gemfire.listener.annotation.ContinuousQuery
* @see org.springframework.util.ErrorHandler
* @since 2.0.0
*/
@Configuration
@@ -83,24 +88,36 @@ public class ContinuousQueryConfiguration extends AbstractAnnotationConfigSuppor
private String errorHandlerBeanName;
private String poolName;
private String queryServiceBeanName;
private String taskExecutorBeanName;
/**
* Returns the {@link Annotation} {@link Class type} that configures and creates {@link CqQuery Continuous Queries}
* for application {@link ContinuousQuery} annotated POJO service methods.
*
* @return the {@link Annotation} {@link Class type} that configures and creates {@link CqQuery Continuous Queries}
* for application {@link ContinuousQuery} annotated POJO service methods.
* @see org.springframework.data.gemfire.config.annotation.EnableContinuousQueries
* @see org.springframework.data.gemfire.listener.annotation.ContinuousQuery
* @see java.lang.annotation.Annotation
* @see java.lang.Class
*/
@Override
protected Class<? extends Annotation> getAnnotationType() {
return EnableContinuousQueries.class;
}
@Override
public void setImportMetadata(AnnotationMetadata importMetadata) {
public void setImportMetadata(AnnotationMetadata importingClassMetadata) {
if (importMetadata.hasAnnotation(getAnnotationType().getName())) {
if (isAnnotationPresent(importingClassMetadata)) {
AnnotationAttributes enableContinuousQueriesAttributes =
AnnotationAttributes.fromMap(importMetadata.getAnnotationAttributes(getAnnotationType().getName()));
AnnotationAttributes enableContinuousQueriesAttributes = getAnnotationAttributes(importingClassMetadata);
setErrorHandlerBeanName(enableContinuousQueriesAttributes.getString("errorHandlerBeanName"));
setPhase(enableContinuousQueriesAttributes.<Integer>getNumber("phase"));
setPoolName(enableContinuousQueriesAttributes.getString("poolName"));
setQueryServiceBeanName(enableContinuousQueriesAttributes.getString("queryServiceBeanName"));
setTaskExecutorBeanName(enableContinuousQueriesAttributes.getString("taskExecutorBeanName"));
}
}
@@ -114,7 +131,7 @@ public class ContinuousQueryConfiguration extends AbstractAnnotationConfigSuppor
private List<ContinuousQueryDefinition> continuousQueryDefinitions = new ArrayList<>();
@Nullable @Override
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof ContinuousQueryListenerContainer) {
@@ -190,6 +207,7 @@ public class ContinuousQueryConfiguration extends AbstractAnnotationConfigSuppor
resolveErrorHandler().ifPresent(container::setErrorHandler);
resolvePhase().ifPresent(container::setPhase);
resolvePoolName().ifPresent(container::setPoolName);
resolveQueryService().ifPresent(container::setQueryService);
resolveTaskExecutor().ifPresent(container::setTaskExecutor);
return container;
@@ -230,6 +248,13 @@ public class ContinuousQueryConfiguration extends AbstractAnnotationConfigSuppor
return Optional.ofNullable(getPoolName()).filter(StringUtils::hasText);
}
protected Optional<QueryService> resolveQueryService() {
return Optional.ofNullable(getQueryServiceBeanName())
.filter(StringUtils::hasText)
.map(queryServiceBeanName -> getBeanFactory().getBean(queryServiceBeanName, QueryService.class));
}
protected Optional<Executor> resolveTaskExecutor() {
return Optional.ofNullable(getTaskExecutorBeanName())
@@ -242,7 +267,7 @@ public class ContinuousQueryConfiguration extends AbstractAnnotationConfigSuppor
}
protected String getErrorHandlerBeanName() {
return errorHandlerBeanName;
return this.errorHandlerBeanName;
}
public void setPhase(int phase) {
@@ -250,7 +275,7 @@ public class ContinuousQueryConfiguration extends AbstractAnnotationConfigSuppor
}
protected int getPhase() {
return phase;
return this.phase;
}
public void setPoolName(String poolName) {
@@ -258,7 +283,15 @@ public class ContinuousQueryConfiguration extends AbstractAnnotationConfigSuppor
}
protected String getPoolName() {
return poolName;
return this.poolName;
}
public void setQueryServiceBeanName(String queryServiceBeanName) {
this.queryServiceBeanName = queryServiceBeanName;
}
protected String getQueryServiceBeanName() {
return this.queryServiceBeanName;
}
public void setTaskExecutorBeanName(String taskExecutorBeanName) {
@@ -266,6 +299,6 @@ public class ContinuousQueryConfiguration extends AbstractAnnotationConfigSuppor
}
protected String getTaskExecutorBeanName() {
return taskExecutorBeanName;
return this.taskExecutorBeanName;
}
}

9
src/main/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueries.java
View File

@@ -25,6 +25,7 @@ import java.lang.annotation.Target;
import java.util.concurrent.Executor;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.query.QueryService;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer;
@@ -41,6 +42,7 @@ import org.springframework.util.ErrorHandler;
* @see java.lang.annotation.Target
* @see java.util.concurrent.Executor
* @see org.apache.geode.cache.client.Pool
* @see org.apache.geode.cache.query.QueryService
* @see org.springframework.context.annotation.Configuration
* @see org.springframework.context.annotation.Import
* @see org.springframework.data.gemfire.config.annotation.ContinuousQueryConfiguration
@@ -78,6 +80,13 @@ public @interface EnableContinuousQueries {
*/
String poolName() default "";
/**
* Refers to the name of the {@link QueryService} bean used to define CQs.
*
* Defaults to unset.
*/
String queryServiceBeanName() default "";
/**
* Refers to the name of the {@link Executor} bean used to process CQ events asynchronously.
*

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

@@ -496,7 +496,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
* @return an {@link Optional} reference to the configured {@link ErrorHandler}.
* @see org.springframework.util.ErrorHandler
*/
protected Optional<ErrorHandler> getErrorHandler() {
public Optional<ErrorHandler> getErrorHandler() {
return Optional.ofNullable(this.errorHandler);
}
@@ -534,7 +534,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
*
* @return the configured {@link String pool name}.
*/
protected String getPoolName() {
public String getPoolName() {
return this.poolName;
}
@@ -564,7 +564,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
* @return a reference to the configured {@link QueryService}.
* @see org.apache.geode.cache.query.QueryService
*/
protected QueryService getQueryService() {
public QueryService getQueryService() {
return this.queryService;
}
@@ -587,7 +587,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
* @return a reference to the configured {@link Executor TaskExecutor}.
* @see java.util.concurrent.Executor
*/
protected Executor getTaskExecutor() {
public Executor getTaskExecutor() {
return this.taskExecutor;
}

72
src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationUnitTests.java
View File

@@ -22,9 +22,11 @@ import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.springframework.data.gemfire.util.ArrayUtils.asArray;
import java.lang.reflect.Proxy;
import java.util.concurrent.Executor;
import org.apache.geode.cache.GemFireCache;
import org.apache.geode.cache.Region;
@@ -61,8 +63,8 @@ import org.springframework.data.gemfire.test.model.Person;
import org.springframework.data.gemfire.test.repo.PersonRepository;
import org.springframework.data.gemfire.test.support.IOUtils;
import org.springframework.data.repository.CrudRepository;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;
import org.springframework.util.ErrorHandler;
import lombok.Data;
@@ -90,6 +92,40 @@ public class EnableContinuousQueriesConfigurationUnitTests {
return new AnnotationConfigApplicationContext(annotatedClasses);
}
@Test
public void continuousQueryListenerContainerConfigurationIsCorrect() {
ConfigurableApplicationContext applicationContext =
newApplicationContext(TestContinuousQueryListenerContainerConfiguration.class);
try {
ErrorHandler mockErrorHandler = applicationContext.getBean("mockErrorHandler", ErrorHandler.class);
Pool mockPool = applicationContext.getBean("mockPool", Pool.class);
QueryService mockQueryService = applicationContext.getBean("mockQueryService", QueryService.class);
Executor mockTaskExecutor = applicationContext.getBean("mockTaskExecutor", Executor.class);
assertThat(applicationContext.containsBean("continuousQueryListenerContainer")).isTrue();
ContinuousQueryListenerContainer container =
applicationContext.getBean("continuousQueryListenerContainer",
ContinuousQueryListenerContainer.class);
assertThat(container).isNotNull();
assertThat(container.getErrorHandler().orElse(null)).isEqualTo(mockErrorHandler);
assertThat(container.getPhase()).isEqualTo(1);
assertThat(container.getPoolName()).isEqualTo(mockPool.getName());
assertThat(container.getQueryService()).isEqualTo(mockQueryService);
assertThat(container.getTaskExecutor()).isEqualTo(mockTaskExecutor);
}
finally {
IOUtils.close(applicationContext);
}
}
private void testRegisterAndExecuteContinuousQuery(Class<?>... annotatedClasses) throws Exception {
ConfigurableApplicationContext applicationContext = newApplicationContext(annotatedClasses);
@@ -134,6 +170,38 @@ public class EnableContinuousQueriesConfigurationUnitTests {
testRegisterAndExecuteContinuousQuery(TestContinuousQueryRegistrationAndExecutionOnProxiedBeanConfiguration.class);
}
@ClientCacheApplication
@EnableGemFireMockObjects
@EnableContinuousQueries(errorHandlerBeanName = "mockErrorHandler", phase = 1, poolName = "mockPool",
queryServiceBeanName = "mockQueryService", taskExecutorBeanName = "mockTaskExecutor")
static class TestContinuousQueryListenerContainerConfiguration {
@Bean
ErrorHandler mockErrorHandler() {
return mock(ErrorHandler.class);
}
@Bean
Pool mockPool() {
Pool mockPool = mock(Pool.class);
when(mockPool.getName()).thenReturn("mockPool");
return mockPool;
}
@Bean
QueryService mockQueryService() {
return mock(QueryService.class);
}
@Bean
Executor mockTaskExecutor() {
return mock(Executor.class);
}
}
@ClientCacheApplication
@EnableContinuousQueries
@EnableGemFireMockObjects
@@ -171,7 +239,7 @@ public class EnableContinuousQueriesConfigurationUnitTests {
return new BeanPostProcessor() {
@Nullable @Override
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
if ("continuousQueryComponent".equals(beanName)) {