From 02d0c8e60aa175a0fdbb8951e7bfd4ec066d4e98 Mon Sep 17 00:00:00 2001 From: John Blum Date: Tue, 21 Nov 2017 23:00:22 -0800 Subject: [PATCH] DATAGEODE-64 - Allow the QueryService to be configured as a Spring bean in the ContinuousQuery Annotation support. --- .../ContinuousQueryConfiguration.java | 55 +++++++++++--- .../annotation/EnableContinuousQueries.java | 9 +++ .../ContinuousQueryListenerContainer.java | 8 +-- ...ntinuousQueriesConfigurationUnitTests.java | 72 ++++++++++++++++++- 4 files changed, 127 insertions(+), 17 deletions(-) diff --git a/src/main/java/org/springframework/data/gemfire/config/annotation/ContinuousQueryConfiguration.java b/src/main/java/org/springframework/data/gemfire/config/annotation/ContinuousQueryConfiguration.java index 1e554b6a..998bcc4b 100644 --- a/src/main/java/org/springframework/data/gemfire/config/annotation/ContinuousQueryConfiguration.java +++ b/src/main/java/org/springframework/data/gemfire/config/annotation/ContinuousQueryConfiguration.java @@ -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 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.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 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 resolveQueryService() { + + return Optional.ofNullable(getQueryServiceBeanName()) + .filter(StringUtils::hasText) + .map(queryServiceBeanName -> getBeanFactory().getBean(queryServiceBeanName, QueryService.class)); + } + protected Optional 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; } } diff --git a/src/main/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueries.java b/src/main/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueries.java index 5157f1c8..e310877c 100644 --- a/src/main/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueries.java +++ b/src/main/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueries.java @@ -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. * diff --git a/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java b/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java index 811e596a..303f2be4 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java +++ b/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java @@ -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 getErrorHandler() { + public Optional 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; } diff --git a/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationUnitTests.java b/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationUnitTests.java index aea9ef0d..fd69c10f 100644 --- a/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationUnitTests.java +++ b/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationUnitTests.java @@ -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)) {