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 new file mode 100644 index 00000000..9753c5c3 --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/config/annotation/ContinuousQueryConfiguration.java @@ -0,0 +1,243 @@ +/* + * Copyright 2017 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.gemfire.config.annotation; + +import static java.util.Arrays.stream; +import static org.springframework.data.gemfire.util.CollectionUtils.nullSafeMap; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.Executor; +import java.util.stream.Collectors; + +import org.apache.geode.cache.GemFireCache; +import org.apache.shiro.util.Assert; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.ImportAware; +import org.springframework.core.annotation.AnnotationAttributes; +import org.springframework.core.type.AnnotationMetadata; +import org.springframework.data.gemfire.config.annotation.support.AbstractAnnotationConfigSupport; +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.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. + * + * @author John Blum + * @see org.apache.geode.cache.GemFireCache + * @see org.springframework.beans.factory.config.BeanPostProcessor + * @see org.springframework.context.annotation.Bean + * @see org.springframework.context.annotation.Configuration + * @see org.springframework.context.annotation.ImportAware + * @see org.springframework.core.annotation.AnnotationAttributes + * @see org.springframework.core.type.AnnotationMetadata + * @see org.springframework.data.gemfire.config.annotation.support.AbstractAnnotationConfigSupport + * @see org.springframework.data.gemfire.listener.ContinuousQueryDefinition + * @see org.springframework.data.gemfire.listener.ContinuousQueryListener + * @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer + * @since 2.0.0 + */ +@Configuration +@SuppressWarnings("unused") +public class ContinuousQueryConfiguration extends AbstractAnnotationConfigSupport implements ImportAware { + + protected static final String ORG_SPRINGFRAMEWORK_DATA_GEMFIRE_PACKAGE_NAME = "org.springframework.data.gemfire"; + protected static final String ORG_SPRINGFRAMEWORK_PACKAGE_NAME = "org.springframework"; + + private int phase; + + @Autowired(required = false) + private List configurers = Collections.emptyList(); + + private String errorHandlerBeanName; + private String poolName; + private String taskExecutorBeanName; + + @Override + protected Class getAnnotationType() { + return EnableContinuousQueries.class; + } + + @Override + public void setImportMetadata(AnnotationMetadata importMetadata) { + + if (importMetadata.hasAnnotation(getAnnotationType().getName())) { + + AnnotationAttributes enableContinuousQueriesAttributes = + AnnotationAttributes.fromMap(importMetadata.getAnnotationAttributes(getAnnotationType().getName())); + + setErrorHandlerBeanName(enableContinuousQueriesAttributes.getString("errorHandlerBeanName")); + setPhase(enableContinuousQueriesAttributes.getNumber("phase")); + setPoolName(enableContinuousQueriesAttributes.getString("poolName")); + setTaskExecutorBeanName(enableContinuousQueriesAttributes.getString("taskExecutorBeanName")); + } + } + + @Bean + public BeanPostProcessor continuousQueryBeanPostProcessor() { + + return new BeanPostProcessor() { + + private ContinuousQueryListenerContainer container; + + private List continuousQueryDefinitions = new ArrayList<>(); + + @Nullable @Override + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + + if (bean instanceof ContinuousQueryListenerContainer) { + this.container = (ContinuousQueryListenerContainer) bean; + this.continuousQueryDefinitions.forEach(definition -> this.container.addListener(definition)); + this.continuousQueryDefinitions.clear(); + } + else if (isApplicationBean(bean)) { + + List definitions = stream(bean.getClass().getMethods()) + .filter(method -> method.isAnnotationPresent(ContinuousQuery.class)) + .map(method -> ContinuousQueryDefinition.from(bean, method)) + .collect(Collectors.toList()); + + Optional.ofNullable(this.container).map(container -> { + definitions.forEach(container::addListener); + return container; + }).orElseGet(() -> { + this.continuousQueryDefinitions.addAll(definitions); + return null; + }); + + } + + return bean; + } + }; + } + + private boolean isApplicationBean(Object bean) { + + return Optional.ofNullable(bean) + .map(Object::getClass) + .filter(type -> type.getPackage().getName().startsWith(ORG_SPRINGFRAMEWORK_DATA_GEMFIRE_PACKAGE_NAME) + || !type.getPackage().getName().startsWith(ORG_SPRINGFRAMEWORK_PACKAGE_NAME)) + .isPresent(); + } + + @Bean + public ContinuousQueryListenerContainer continuousQueryListenerContainer(GemFireCache gemfireCache) { + + Assert.state(CacheUtils.isClient(gemfireCache), + "Continuous Queries (CQ) may only be used in a ClientCache application"); + + ContinuousQueryListenerContainer container = new ContinuousQueryListenerContainer(); + + container.setCache(gemfireCache); + container.setContinuousQueryListenerContainerConfigurers(resolveContinuousQueryListenerContainerConfigurers()); + + resolveErrorHandler().ifPresent(container::setErrorHandler); + resolvePhase().ifPresent(container::setPhase); + resolvePoolName().ifPresent(container::setPoolName); + resolveTaskExecutor().ifPresent(container::setTaskExecutor); + + return container; + } + + protected List resolveContinuousQueryListenerContainerConfigurers() { + + return Optional.ofNullable(this.configurers) + .filter(configurers -> !configurers.isEmpty()) + .orElseGet(() -> + Optional.of(this.beanFactory()) + .filter(beanFactory -> beanFactory instanceof ListableBeanFactory) + .map(beanFactory -> { + + Map beansOfType = + ((ListableBeanFactory) beanFactory).getBeansOfType(ContinuousQueryListenerContainerConfigurer.class, + true, true); + + return nullSafeMap(beansOfType).values().stream().collect(Collectors.toList()); + }) + .orElseGet(Collections::emptyList) + ); + } + + protected Optional resolveErrorHandler() { + + return Optional.ofNullable(getErrorHandlerBeanName()) + .filter(StringUtils::hasText) + .map(errorHandlerBeanName -> beanFactory().getBean(errorHandlerBeanName, ErrorHandler.class)); + } + + protected Optional resolvePhase() { + return Optional.of(getPhase()).filter(phase -> phase != 0); + } + + protected Optional resolvePoolName() { + return Optional.ofNullable(getPoolName()).filter(StringUtils::hasText); + } + + protected Optional resolveTaskExecutor() { + + return Optional.ofNullable(getTaskExecutorBeanName()) + .filter(StringUtils::hasText) + .map(taskExecutorBeanName -> beanFactory().getBean(taskExecutorBeanName, Executor.class)); + } + + public void setErrorHandlerBeanName(String errorHandlerBeanName) { + this.errorHandlerBeanName = errorHandlerBeanName; + } + + protected String getErrorHandlerBeanName() { + return errorHandlerBeanName; + } + + public void setPhase(int phase) { + this.phase = phase; + } + + protected int getPhase() { + return phase; + } + + public void setPoolName(String poolName) { + this.poolName = poolName; + } + + protected String getPoolName() { + return poolName; + } + + public void setTaskExecutorBeanName(String taskExecutorBeanName) { + this.taskExecutorBeanName = taskExecutorBeanName; + } + + protected String getTaskExecutorBeanName() { + return taskExecutorBeanName; + } +} diff --git a/src/main/java/org/springframework/data/gemfire/config/annotation/ContinuousQueryListenerContainerConfigurer.java b/src/main/java/org/springframework/data/gemfire/config/annotation/ContinuousQueryListenerContainerConfigurer.java new file mode 100644 index 00000000..d06f4b21 --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/config/annotation/ContinuousQueryListenerContainerConfigurer.java @@ -0,0 +1,41 @@ +/* + * Copyright 2017 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.gemfire.config.annotation; + +import org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer; + +/** + * The {@link ContinuousQueryListenerContainerConfigurer} interfaces defines a contract for implementations to customize + * the configuration of SDG's {@link ContinuousQueryListenerContainer} when enabling Continuous Query (CQ) functionality + * in Spring Boot, GemFire/Geode cache client applications. + * + * @author John Blum + * @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer + * @since 2.0.0 + */ +public interface ContinuousQueryListenerContainerConfigurer { + + /** + * Applies addditional configuration to the declared/defined {@link ContinuousQueryListenerContainer}. + * + * @param beanName {@link String name} of the {@link ContinuousQueryListenerContainer} bean definition. + * @param container reference to the {@link ContinuousQueryListenerContainer} instance. + * @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer + */ + void configure(String beanName, ContinuousQueryListenerContainer container); + +} 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 new file mode 100644 index 00000000..17b185d6 --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueries.java @@ -0,0 +1,89 @@ +/* + * Copyright 2017 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.gemfire.config.annotation; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Inherited; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; +import java.util.concurrent.Executor; + +import org.apache.geode.cache.client.Pool; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.data.gemfire.config.xml.GemfireConstants; +import org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer; +import org.springframework.util.ErrorHandler; + +/** + * The {@link EnableContinuousQueries} annotation marks a Spring {@link Configuration @Configuration} annotated + * application configuration class to enable Pivotal GemFire / Apache Geode Continuous Queries (CQ) feature. + * + * @author John Blum + * @see java.lang.annotation.Documented + * @see java.lang.annotation.Inherited + * @see java.lang.annotation.Retention + * @see java.lang.annotation.Target + * @see java.util.concurrent.Executor + * @see org.apache.geode.cache.client.Pool + * @see org.springframework.context.annotation.Configuration + * @see org.springframework.context.annotation.Import + * @see org.springframework.data.gemfire.config.annotation.ContinuousQueryConfiguration + * @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer + * @since @.0.0 + */ +@Target(ElementType.TYPE) +@Retention(RetentionPolicy.RUNTIME) +@Inherited +@Documented +@Import(ContinuousQueryConfiguration.class) +@SuppressWarnings("unused") +public @interface EnableContinuousQueries { + + /** + * Refers to the {@link String name} of the declared {@link ErrorHandler} bean that will handle errors + * thrown during CQ event processing by CQ listeners. + * + * Defaults to unset. + */ + String errorHandlerBeanName() default ""; + + /** + * Defines the Spring container lifecycle phase in which the SDG {@link ContinuousQueryListenerContainer} + * will be started on auto-start. + * + * Defaults to {@literal 0}. + */ + int phase() default 0; + + /** + * Refers to the name of the {@link Pool} over which CQs are registered and CQ events are received. + * + * Defaults to {@literal gemfirePool}. + */ + String poolName() default GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME; + + /** + * Refers to the name of the {@link Executor} bean used to process CQ events asynchronously. + * + * Defaults to unset. + */ + String taskExecutorBeanName() default ""; + +} diff --git a/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryDefinition.java b/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryDefinition.java index e8ecf6b6..5c398036 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryDefinition.java +++ b/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryDefinition.java @@ -16,6 +16,8 @@ package org.springframework.data.gemfire.listener; +import java.lang.reflect.Method; +import java.util.Optional; import java.util.function.Function; import org.apache.geode.cache.query.CqAttributes; @@ -24,6 +26,8 @@ import org.apache.geode.cache.query.CqListener; import org.apache.geode.cache.query.CqQuery; import org.apache.shiro.util.StringUtils; import org.springframework.beans.factory.InitializingBean; +import org.springframework.data.gemfire.listener.adapter.ContinuousQueryListenerAdapter; +import org.springframework.data.gemfire.listener.annotation.ContinuousQuery; import org.springframework.util.Assert; /** @@ -43,6 +47,30 @@ public class ContinuousQueryDefinition implements InitializingBean { private final String name; private final String query; + public static ContinuousQueryDefinition from(Object delegate, Method method) { + + Assert.notNull(method, "Method must not be null"); + + ContinuousQuery continuousQuery = method.getAnnotation(ContinuousQuery.class); + + Assert.notNull(continuousQuery, () -> String.format("Method [%1$s] must be annotated with [%2$s]", + method, ContinuousQuery.class.getName())); + + String name = Optional.of(continuousQuery.name()) + .filter(org.springframework.util.StringUtils::hasText) + .orElseGet(() -> String.format("%1$s.%2$s", method.getDeclaringClass().getName(), method.getName())); + + String query = continuousQuery.query(); + + ContinuousQueryListenerAdapter listener = new ContinuousQueryListenerAdapter(delegate); + + listener.setDefaultListenerMethod(method.getName()); + + boolean durable = continuousQuery.durable(); + + return new ContinuousQueryDefinition(name, query, listener, durable); + } + public ContinuousQueryDefinition(String query, ContinuousQueryListener listener) { this(query, listener, false); } 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 b795eee1..11111c90 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java +++ b/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java @@ -16,10 +16,17 @@ 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; import java.util.LinkedHashSet; +import java.util.List; import java.util.Optional; import java.util.Queue; import java.util.Set; @@ -52,6 +59,7 @@ import org.springframework.data.gemfire.GemfireQueryException; import org.springframework.data.gemfire.GemfireUtils; import org.springframework.data.gemfire.client.support.DefaultableDelegatingPoolAdapter; import org.springframework.data.gemfire.client.support.DelegatingPoolAdapter; +import org.springframework.data.gemfire.config.annotation.ContinuousQueryListenerContainerConfigurer; import org.springframework.data.gemfire.config.xml.GemfireConstants; import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; @@ -82,13 +90,13 @@ import org.springframework.util.StringUtils; * @see org.springframework.data.gemfire.client.support.DefaultableDelegatingPoolAdapter * @see org.springframework.data.gemfire.client.support.DelegatingPoolAdapter * @see org.springframework.util.ErrorHandler - * @since 1.1 + * @since 1.1.0 */ @SuppressWarnings("unused") public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanNameAware, InitializingBean, DisposableBean, SmartLifecycle { - // Default Thread name prefix is "ContinuousQueryListenerContainer-". + // Default Thread name prefix is "ContinuousQueryListenerContainer-" public static final String DEFAULT_THREAD_NAME_PREFIX = String.format("%s-", ContinuousQueryListenerContainer.class.getSimpleName()); @@ -106,6 +114,12 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN private Executor taskExecutor; + private List cqListenerContainerConfigurers = Collections.emptyList(); + + private ContinuousQueryListenerContainerConfigurer compositeCqListenerContainerConfigurer = + (beanName, container) -> nullSafeList(cqListenerContainerConfigurers).forEach(configurer -> + configurer.configure(beanName, container)); + protected final Log logger = LogFactory.getLog(getClass()); private Queue continuousQueries = new ConcurrentLinkedQueue<>(); @@ -120,6 +134,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN @Override public void afterPropertiesSet() { + applyContinuousQueryListenerContainerConfigurers(); validateQueryService(initQueryService(eagerlyInitializePool(resolvePoolName()))); initExecutor(); initContinuousQueries(); @@ -127,6 +142,11 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN this.initialized = true; } + /* (non-Javadoc) */ + private void applyContinuousQueryListenerContainerConfigurers() { + applyContinuousQueryListenerContainerConfigurers(getCompositeContinuousQueryListenerContainerConfigurer()); + } + /* (non-Javadoc) */ private QueryService validateQueryService(QueryService queryService) { @@ -135,6 +155,36 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN return queryService; } + /** + * Applies the array of {@link ContinuousQueryListenerContainerConfigurer} objects to customize the configuration + * of this {@link ContinuousQueryListenerContainer}. + * + * @param configurers array of {@link ContinuousQueryListenerContainerConfigurer} used to customize + * the configuration of this {@link ContinuousQueryListenerContainer}. + * @see org.springframework.data.gemfire.config.annotation.ContinuousQueryListenerContainerConfigurer + */ + protected void applyContinuousQueryListenerContainerConfigurers( + ContinuousQueryListenerContainerConfigurer... configurers) { + + applyContinuousQueryListenerContainerConfigurers(Arrays.asList( + nullSafeArray(configurers, ContinuousQueryListenerContainerConfigurer.class))); + } + + /** + * Applies the {@link Iterable} of {@link ContinuousQueryListenerContainerConfigurer} objects to customize + * the configuration of this {@link ContinuousQueryListenerContainer}. + * + * @param configurers {@link Iterable} of {@link ContinuousQueryListenerContainerConfigurer} used to customize + * the configuration of this {@link ContinuousQueryListenerContainer}. + * @see org.springframework.data.gemfire.config.annotation.ContinuousQueryListenerContainerConfigurer + */ + protected void applyContinuousQueryListenerContainerConfigurers( + Iterable configurers) { + + stream(nullSafeIterable(configurers).spliterator(), false) + .forEach(configurer -> configurer.configure(getBeanName(), this)); + } + /** * Resolves the name of the {@link Pool} configured to handle the registered Continuous Queries. * @@ -175,6 +225,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN }; return Optional.ofNullable(getBeanFactory()) + .filter(it -> it.containsBean(poolName)) .filter(it -> it.isTypeMatch(poolName, Pool.class)) .map(it -> { try { @@ -374,6 +425,45 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN return this.continuousQueries; } + /** + * Null-safe operation setting an array of {@link ContinuousQueryListenerContainerConfigurer} objects used to + * customize the configuration of this {@link ContinuousQueryListenerContainer}. + * + * @param configurers array of {@link ContinuousQueryListenerContainerConfigurer} objects used to customize + * the configuration of this {@link ContinuousQueryListenerContainer}. + * @see org.springframework.data.gemfire.config.annotation.ContinuousQueryListenerContainerConfigurer + * @see #setContinuousQueryListenerContainerConfigurers(List) + */ + public void setContinuousQueryListenerContainerConfigurers(ContinuousQueryListenerContainerConfigurer... configurers) { + setContinuousQueryListenerContainerConfigurers(Arrays.asList( + nullSafeArray(configurers, ContinuousQueryListenerContainerConfigurer.class))); + } + + /** + * Null-safe operation setting an {@link Iterable} of {@link ContinuousQueryListenerContainerConfigurer} objects + * used to customize the configuration of this {@link ContinuousQueryListenerContainer}. + * + * @param configurers {@link Iterable} of {@link ContinuousQueryListenerContainerConfigurer} objects used to + * customize the configuration of this {@link ContinuousQueryListenerContainer}. + * @see org.springframework.data.gemfire.config.annotation.ContinuousQueryListenerContainerConfigurer + */ + public void setContinuousQueryListenerContainerConfigurers(List configurers) { + this.cqListenerContainerConfigurers = Optional.ofNullable(configurers).orElseGet(Collections::emptyList); + } + + /** + * Returns a Composite object containing + * the collection of {@link ContinuousQueryListenerContainerConfigurer} objects used to customize the configuration + * of this {@link ContinuousQueryListenerContainer}. + * + * @return a Composite object containing a collection of {@link ContinuousQueryListenerContainerConfigurer} objects + * used to customize the configuration of this {@link ContinuousQueryListenerContainer}. + * @see org.springframework.data.gemfire.config.annotation.ContinuousQueryListenerContainerConfigurer + */ + protected ContinuousQueryListenerContainerConfigurer getCompositeContinuousQueryListenerContainerConfigurer() { + return this.compositeCqListenerContainerConfigurer; + } + /** * Set an {@link ErrorHandler} to be invoked in case of any uncaught {@link Exception Exceptions} thrown * while processing a CQ event. diff --git a/src/main/java/org/springframework/data/gemfire/listener/adapter/ContinuousQueryListenerAdapter.java b/src/main/java/org/springframework/data/gemfire/listener/adapter/ContinuousQueryListenerAdapter.java index dcb6d40f..732c3471 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/adapter/ContinuousQueryListenerAdapter.java +++ b/src/main/java/org/springframework/data/gemfire/listener/adapter/ContinuousQueryListenerAdapter.java @@ -21,6 +21,7 @@ import java.lang.reflect.Method; import java.lang.reflect.Modifier; import java.util.ArrayList; import java.util.List; +import java.util.Optional; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -33,34 +34,31 @@ import org.springframework.data.gemfire.listener.ContinuousQueryListener; import org.springframework.data.gemfire.listener.GemfireListenerExecutionFailedException; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; -import org.springframework.util.ReflectionUtils.MethodCallback; -import org.springframework.util.ReflectionUtils.MethodFilter; +import org.springframework.util.StringUtils; /** - * Event listener adapter that delegates the handling of messages to target - * listener methods via reflection, with flexible event type conversion. - * Allows listener methods to operate on event content types, completely - * independent from the GemFire API. + * Event listener adapter that delegates the handling of messages to target listener methods via reflection, + * with flexible event type conversion. * - *

Modeled as much as possible after the JMS MessageListenerAdapter in - * Spring Framework. + * Allows listener methods to operate on event content types, completely independent from the GemFire/Geode API. * - *

By default, the content of incoming GemFire events gets extracted before - * being passed into the target listener method, to let the target method - * operate on event content types such as Object or Operation instead of - * the raw {@link CqEvent}.

+ *

Modeled as much as possible after the JMS MessageListenerAdapter in the core Spring Framework. * - *

Find below some examples of method signatures compliant with this - * adapter class. This first example handles all CqEvent types - * and gets passed the contents of each event type as an - * argument.

+ *

By default, the content of incoming GemFire/Geode CQ events gets extracted before being passed into + * the target listener method, to let the target method operate on event content types such as Object or Operation + * instead of the raw {@link CqEvent}.

+ * + *

Find below some examples of method signatures compliant with this adapter class. + * + * This first example handles all CqEvent types and gets passed the contents of each + * event type as an argument.

* * 
public interface PojoListener {
  *    void handleEvent(CqEvent event);
  *    void handleEvent(Operation baseOp);
  *    void handleEvent(Object key);
  *    void handleEvent(Object key, Object newValue);
- *    void handleEvent(Throwable th);
+ *    void handleEvent(Throwable cause);
  *    void handleEvent(CqEvent event, Operation baseOp, byte[] deltaValue);
  *    void handleEvent(CqEvent event, Operation baseOp, Operation queryOp, Object key, Object newValue);
  * }
@@ -69,6 +67,12 @@ import org.springframework.util.ReflectionUtils.MethodFilter; * @author Costin Leau * @author Oliver Gierke * @author John Blum + * @see java.lang.reflect.Method + * @see org.apache.geode.cache.Operation + * @see org.apache.geode.cache.query.CqEvent + * @see org.apache.geode.cache.query.CqQuery + * @see org.springframework.data.gemfire.listener.ContinuousQueryListener + * @since 1.1.0 */ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { @@ -84,40 +88,45 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { private String defaultListenerMethod = DEFAULT_LISTENER_METHOD_NAME; /** - * Create a new {@link ContinuousQueryListenerAdapter} with default settings. + * Constructs a new instance of {@link ContinuousQueryListenerAdapter} with default settings. */ public ContinuousQueryListenerAdapter() { setDelegate(this); } /** - * Create a new {@link ContinuousQueryListenerAdapter} for the given delegate. + * Constructs a new instance of {@link ContinuousQueryListenerAdapter} initialized with + * the given {@link Object delegate}. * - * @param delegate the delegate object + * @param delegate {@link Object delegate} of the listener method event callback. + * @see #setDelegate(Object) */ public ContinuousQueryListenerAdapter(Object delegate) { setDelegate(delegate); } /** - * Set a target object to delegate events listening to. - * Specified listener methods have to be present on this target object. - *

If no explicit delegate object has been specified, listener - * methods are expected to present on this adapter instance, that is, - * on a custom subclass of this adapter, defining listener methods. + * Sets the target object to which CQ events are delegated. * - * @param delegate delegate object + * Specified listener methods have to be present on this target object. If no explicit delegate object + * has been configured then listener methods are expected to present on this adapter instance. + * In other words, on a custom subclass of this adapter, defining listener methods. + * + * @param delegate {@link Object} to delegate listening for CQ events. + * @throws IllegalArgumentException if {@link Object delegate} is {@literal null}. */ - public void setDelegate(Object delegate) { - Assert.notNull(delegate, "'delegate' must not be null"); + public final void setDelegate(Object delegate) { + + Assert.notNull(delegate, "Delegate is required"); + this.delegate = delegate; this.invoker = null; } /** - * Returns the target object to delegate event listening to. + * Returns a reference to the target object used to listen for and handle CQ events. * - * @return event listening delegation + * @return a reference to the target object used to listen for and handle CQ events. */ public Object getDelegate() { return this.delegate; @@ -144,43 +153,6 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { return this.defaultListenerMethod; } - /** - * Standard {@link ContinuousQueryListener} entry point. - *

Delegates the event to the target listener method, with appropriate - * conversion of the event argument. In case of an exception, the - * {@link #handleListenerException(Throwable)} method will be invoked. - * - * @param event the incoming GemFire event - * @see #handleListenerException - */ - public void onEvent(CqEvent event) { - try { - // Check whether the delegate is a ContinuousQueryListener implementation itself. - // If so, this adapter will simply act as a pass-through. - if (delegate != this && delegate instanceof ContinuousQueryListener) { - ((ContinuousQueryListener) delegate).onEvent(event); - } - // Else... find the listener handler method reflectively. - else { - String methodName = getListenerMethodName(event); - - if (methodName == null) { - throw new InvalidDataAccessApiUsageException("No default listener method specified." - + " Either specify a non-null value for the 'defaultListenerMethod' property" - + " or override the 'getListenerMethodName' method."); - } - - invoker = (invoker != null ? invoker : new MethodInvoker(delegate, methodName)); - - invokeListenerMethod(event, methodName); - } - - } - catch (Throwable cause) { - handleListenerException(cause); - } - } - /** * Determine the name of the listener method that is supposed to * handle the given event. @@ -195,13 +167,52 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { return getDefaultListenerMethod(); } + /** + * Standard {@link ContinuousQueryListener} callback method for handling CQ events. + * + *

Delegates the CQ event to the target listener method, with appropriate conversion of the event arguments. + * In case of an exception, the {@link #handleListenerException(Throwable)} method will be invoked. + * + * @param event incoming {@link CqEvent CQ event}. + * @see #handleListenerException + */ + @Override + public void onEvent(CqEvent event) { + + try { + // Determine whether the delegate is a ContinuousQueryListener implementation; + // If so, this adapter will simply act as a pass-through + if (this.delegate != this && this.delegate instanceof ContinuousQueryListener) { + ((ContinuousQueryListener) this.delegate).onEvent(event); + } + // Else, find the listener method handler reflectively + else { + + String methodName = Optional.ofNullable(getListenerMethodName(event)) + .filter(StringUtils::hasText) + .orElseThrow(() -> new InvalidDataAccessApiUsageException("No default listener method specified;" + + " Either specify a non-null value for the 'defaultListenerMethod' property" + + " or override the 'getListenerMethodName' method.")); + + this.invoker = Optional.ofNullable(this.invoker) + .orElseGet(() -> new MethodInvoker(this.delegate, methodName)); + + invokeListenerMethod(event, methodName); + } + + } + catch (Throwable cause) { + handleListenerException(cause); + } + } + /** * Handle the given exception that arose during listener execution. * The default implementation logs the exception at error level. * @param cause the exception to handle */ protected void handleListenerException(Throwable cause) { - logger.error("Listener execution failed...", cause); + logger.error("Listener method execution failed", cause); } /** @@ -212,20 +223,20 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { */ protected void invokeListenerMethod(CqEvent event, String methodName) { try { - invoker.invoke(event); + this.invoker.invoke(event); } - catch (InvocationTargetException e) { - if (e.getTargetException() instanceof DataAccessException) { - throw (DataAccessException) e.getTargetException(); + catch (InvocationTargetException cause) { + if (cause.getTargetException() instanceof DataAccessException) { + throw (DataAccessException) cause.getTargetException(); } else { throw new GemfireListenerExecutionFailedException( - String.format("Listener method [%1$s] threw Exception...", methodName), e.getTargetException()); + String.format("Listener method [%s] threw Exception...", methodName), cause.getTargetException()); } } - catch (Throwable e) { + catch (Throwable cause) { throw new GemfireListenerExecutionFailedException( - String.format("Failed to invoke the target listener method [%1$s]", methodName), e); + String.format("Failed to invoke the target listener method [%s]", methodName), cause); } } @@ -233,33 +244,30 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { private final Object delegate; - List methods; + private final List methods; - MethodInvoker(Object delegate, final String methodName) { - Class c = delegate.getClass(); + MethodInvoker(Object delegate, String methodName) { + + Class delegateType = delegate.getClass(); this.delegate = delegate; - methods = new ArrayList(); + this.methods = new ArrayList<>(); - ReflectionUtils.doWithMethods(c, new MethodCallback() { - public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { - ReflectionUtils.makeAccessible(method); - methods.add(method); - } - }, new MethodFilter() { - public boolean matches(Method method) { - return isValidEventMethodSignature(method, methodName); - } - }); + ReflectionUtils.doWithMethods(delegateType, method -> { + ReflectionUtils.makeAccessible(method); + this.methods.add(method); + }, method -> isValidEventMethodSignature(method, methodName)); - Assert.isTrue(!methods.isEmpty(), String.format( - "Cannot find a suitable method named [%1$s#%2$s] - is the method public and does it have the proper arguments?", - c.getName(), methodName)); + Assert.isTrue(!this.methods.isEmpty(), String.format("Cannot find a suitable method named [%1$s#%2$s];" + + " Is the method public and does it have the proper arguments?", + delegateType.getName(), methodName)); } @SuppressWarnings("all") - boolean isValidEventMethodSignature(Method method, String methodName) { - if (Modifier.isPublic(method.getModifiers()) && methodName.equals(method.getName())) { + private boolean isValidEventMethodSignature(Method method, String methodName) { + + if (isEventHandlerMethod(method, methodName)) { + Class[] parameterTypes = method.getParameterTypes(); int objects = 0; @@ -297,20 +305,32 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { return false; } - void invoke(CqEvent event) throws InvocationTargetException, IllegalAccessException { - for (Method method : methods) { - method.invoke(delegate, getMethodArguments(method, event)); + private boolean isEventHandlerMethod(Method method, String methodName) { + + return Optional.ofNullable(method) + .filter(it -> Modifier.isPublic(it.getModifiers())) + .filter(it -> method.getName().equals(methodName)) + .isPresent(); + } + + void invoke(CqEvent event) throws IllegalAccessException, InvocationTargetException { + + for (Method method : this.methods) { + method.invoke(this.delegate, getMethodArguments(method, event)); } } - Object[] getMethodArguments(Method method, CqEvent event) { + private Object[] getMethodArguments(Method method, CqEvent event) { + Class[] parameterTypes = method.getParameterTypes(); + Object[] args = new Object[parameterTypes.length]; boolean query = false; boolean value = false; for (int index = 0; index < parameterTypes.length; index++) { + Class parameterType = parameterTypes[index]; if (Object.class.equals(parameterType)) { @@ -338,5 +358,4 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener { return args; } } - } diff --git a/src/main/java/org/springframework/data/gemfire/listener/annotation/ContinuousQuery.java b/src/main/java/org/springframework/data/gemfire/listener/annotation/ContinuousQuery.java new file mode 100644 index 00000000..fb84ce08 --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/listener/annotation/ContinuousQuery.java @@ -0,0 +1,62 @@ +/* + * Copyright 2017 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.gemfire.listener.annotation; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Inherited; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * The {@link ContinuousQuery} annotation to define a GemFire/Geode Continuous Query (CQ) on a POJO method + * which handles all CQ events and errors. + * + * @author John Blum + * @see java.lang.annotation.Documented + * @see java.lang.annotation.Inherited + * @see java.lang.annotation.Retention + * @see java.lang.annotation.Target + * @since 2.0.0 + */ +@Target(ElementType.METHOD) +@Retention(RetentionPolicy.RUNTIME) +@Inherited +@Documented +public @interface ContinuousQuery { + + /** + * Determines whether the CQ is durable. + * + * Defaults to {@literal false}. + */ + boolean durable() default false; + + /** + * {@link String Name} assigned to the registered CQ. + * + * Defaults to the fully-qualified method name. + */ + String name() default ""; + + /** + * Defines the OQL query used by the CQ to determine CQ events. + */ + String query(); + +} diff --git a/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationIntegrationTests.java b/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationIntegrationTests.java new file mode 100644 index 00000000..9ed4cb82 --- /dev/null +++ b/src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationIntegrationTests.java @@ -0,0 +1,295 @@ +/* + * Copyright 2017 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.gemfire.config.annotation; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.data.gemfire.util.ArrayUtils.asArray; + +import java.io.Serializable; +import java.util.Collections; +import java.util.concurrent.atomic.AtomicInteger; + +import javax.annotation.Resource; + +import org.apache.geode.cache.CacheListener; +import org.apache.geode.cache.CacheLoader; +import org.apache.geode.cache.CacheLoaderException; +import org.apache.geode.cache.EntryEvent; +import org.apache.geode.cache.GemFireCache; +import org.apache.geode.cache.LoaderHelper; +import org.apache.geode.cache.Operation; +import org.apache.geode.cache.Region; +import org.apache.geode.cache.client.ClientRegionShortcut; +import org.apache.geode.cache.query.CqEvent; +import org.apache.geode.cache.util.CacheListenerAdapter; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.data.gemfire.PartitionedRegionFactoryBean; +import org.springframework.data.gemfire.client.ClientRegionFactoryBean; +import org.springframework.data.gemfire.listener.annotation.ContinuousQuery; +import org.springframework.data.gemfire.process.ProcessWrapper; +import org.springframework.data.gemfire.support.ConnectionEndpoint; +import org.springframework.data.gemfire.test.support.ClientServerIntegrationTestsSupport; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringRunner; + +import lombok.Data; +import lombok.NonNull; +import lombok.RequiredArgsConstructor; + +/** + * Integration tests for {@link EnableContinuousQueries}, {@link ContinuousQueryConfiguration} + * and {@link ContinuousQuery}. + * + * @author John Blum + * @see org.junit.Test + * @see org.apache.geode.cache.GemFireCache + * @see org.apache.geode.cache.Region + * @see org.apache.geode.cache.query.CqEvent + * @see org.springframework.data.gemfire.config.annotation.ContinuousQueryConfiguration + * @see org.springframework.data.gemfire.config.annotation.EnableContinuousQueries + * @see org.springframework.data.gemfire.listener.annotation.ContinuousQuery + * @see org.springframework.data.gemfire.process.ProcessWrapper + * @see org.springframework.data.gemfire.test.support.ClientServerIntegrationTestsSupport + * @see org.springframework.test.context.ContextConfiguration + * @see org.springframework.test.context.junit4.SpringRunner + * @since 2.0.0 + */ +@RunWith(SpringRunner.class) +@ContextConfiguration(classes = EnableContinuousQueriesConfigurationIntegrationTests.TestConfiguration.class) +@SuppressWarnings("unused") +public class EnableContinuousQueriesConfigurationIntegrationTests extends ClientServerIntegrationTestsSupport { + + private static AtomicInteger boilingTemperatureReadingsCounter = new AtomicInteger(0); + private static AtomicInteger freezingTemperatureReadingsCounter = new AtomicInteger(0); + private static AtomicInteger totalTemperatureReadingsCounter = new AtomicInteger(0); + + private static ProcessWrapper gemfireServer; + + @BeforeClass + public static void startGemFireServer() throws Exception { + + int availablePort = findAvailablePort(); + + gemfireServer = run(GemFireServerConfiguration.class, + String.format("-D%s=%d", GEMFIRE_CACHE_SERVER_PORT_PROPERTY, availablePort)); + + waitForServerToStart("localhost", availablePort); + + System.setProperty(GEMFIRE_CACHE_SERVER_PORT_PROPERTY, String.valueOf(availablePort)); + } + + @AfterClass + public static void stopGemFireServer() { + System.clearProperty(GEMFIRE_CACHE_SERVER_PORT_PROPERTY); + stop(gemfireServer); + } + + @Resource(name = "TemperatureReadings") + private Region temperatureReadings; + + @Before + public void setup() { + + TemperatureReading temperatureReading = this.temperatureReadings.get(1L); + + assertThat(temperatureReading).isNotNull(); + assertThat(temperatureReading.getTemperature()).isEqualTo(99); + + //System.err.printf("Number of Temperature Readings [%d] on Server [%d]%n", + // this.temperatureReadings.size(), this.temperatureReadings.sizeOnServer()); + + assertThat(this.temperatureReadings.sizeOnServer()).isEqualTo(10); + + //waitOn(() -> totalTemperatureReadingsCounter.get() >= 5, 100L); + + //assertThat(totalTemperatureReadingsCounter.get()).isNotZero(); + + //System.err.printf("Total Temperature Readings [%d]%n", totalTemperatureReadingsCounter.get()); + } + + @Test + public void boilingTemperatureReadingsEqualsThree() { + assertThat(boilingTemperatureReadingsCounter.get()).isEqualTo(3); + } + + @Test + public void freezingTemperatureReadingsEqualsTwo() { + assertThat(freezingTemperatureReadingsCounter.get()).isEqualTo(2); + } + + @Configuration + @EnableContinuousQueries(poolName = "DEFAULT") + @Import(GemFireClientConfiguration.class) + static class TestConfiguration { + + @Bean + TemperatureReadingQueryListeners temperatureReadingQueryListeners() { + return new TemperatureReadingQueryListeners(); + } + } + + static class TemperatureReadingQueryListeners { + + @ContinuousQuery(name = "BoilingTemperatures", query = "SELECT * FROM /TemperatureReadings r WHERE r.temperature >= 212") + public void boilingTemperatures(CqEvent event) { + boilingTemperatureReadingsCounter.incrementAndGet(); + } + + @ContinuousQuery(name = "FreezingTemperatures", query = "SELECT * FROM /TemperatureReadings r WHERE r.temperature <= 32") + public void freezingTemperatures(CqEvent event) { + freezingTemperatureReadingsCounter.incrementAndGet(); + } + } + + @ClientCacheApplication(logLevel = "warning", subscriptionEnabled = true) + static class GemFireClientConfiguration { + + @Bean + ClientCacheConfigurer clientCachePoolPortConfigurer( + @Value("${" + GEMFIRE_CACHE_SERVER_PORT_PROPERTY + ":40404}") int port) { + + return (bean, clientCacheFactoryBean) -> clientCacheFactoryBean.setServers( + Collections.singletonList(new ConnectionEndpoint("localhost", port))); + } + + @Bean(name = "TemperatureReadings") + ClientRegionFactoryBean temperatureReadingsRegion(GemFireCache gemfireCache) { + + ClientRegionFactoryBean temperatureReadings = + new ClientRegionFactoryBean<>(); + + temperatureReadings.setCache(gemfireCache); + temperatureReadings.setCacheListeners(asArray(temperatureReadingCounterListener())); + temperatureReadings.setClose(false); + temperatureReadings.setShortcut(ClientRegionShortcut.PROXY); + + return temperatureReadings; + } + + private CacheListener temperatureReadingCounterListener() { + + return new CacheListenerAdapter() { + + @Override + public void afterCreate(EntryEvent event) { + if (Operation.LOCAL_LOAD_CREATE.equals(event.getOperation())) { + totalTemperatureReadingsCounter.incrementAndGet(); + } + } + }; + } + } + + @CacheServerApplication(name = "EnableContinuousQueriesConfigurationIntegrationTests", logLevel = "warning") + static class GemFireServerConfiguration { + + public static void main(String[] args) { + + AnnotationConfigApplicationContext applicationContext = + new AnnotationConfigApplicationContext(GemFireServerConfiguration.class); + + applicationContext.registerShutdownHook(); + } + + @Bean + CacheServerConfigurer cacheServerPortConfigurer( + @Value("${" + GEMFIRE_CACHE_SERVER_PORT_PROPERTY + ":40404}") int port) { + + return (bean, cacheServerFactoryBean) -> cacheServerFactoryBean.setPort(port); + } + + @Bean(name = "TemperatureReadings") + PartitionedRegionFactoryBean temperatureReadingsRegion(GemFireCache gemfireCache) { + + PartitionedRegionFactoryBean temperatureReadings = + new PartitionedRegionFactoryBean<>(); + + temperatureReadings.setCache(gemfireCache); + temperatureReadings.setCacheLoader(temperatureReadingsLoader()); + temperatureReadings.setClose(false); + temperatureReadings.setPersistent(false); + + return temperatureReadings; + } + + private CacheLoader temperatureReadingsLoader() { + + return new CacheLoader() { + + @Override + public TemperatureReading load(LoaderHelper helper) throws CacheLoaderException { + + long key = helper.getKey(); + + Region temperatureReadings = helper.getRegion(); + + recordTemperature(temperatureReadings, ++key, 213); + recordTemperature(temperatureReadings, ++key, 72); + recordTemperature(temperatureReadings, ++key, 400); + recordTemperature(temperatureReadings, ++key, 1024); + recordTemperature(temperatureReadings, ++key, 43); + recordTemperature(temperatureReadings, ++key, 0); + recordTemperature(temperatureReadings, ++key, 33); + recordTemperature(temperatureReadings, ++key, -45); + recordTemperature(temperatureReadings, ++key, 67); + + return TemperatureReading.newTemperatureReading(99); + } + + private void recordTemperature(Region temperatureReadings, + long key, int temperature) { + + sleep(50); + temperatureReadings.put(key, TemperatureReading.newTemperatureReading(temperature)); + } + + private void sleep(long milliseconds) { + try { + Thread.sleep(milliseconds); + } + catch (InterruptedException ignore) { + } + } + @Override + public void close() { + } + }; + } + } + + @Data + @RequiredArgsConstructor(staticName = "newTemperatureReading") + public static class TemperatureReading implements Serializable { + + @NonNull + private Integer temperature; + + @Override + public String toString() { + return String.format("%d °F", getTemperature()); + } + } +} diff --git a/src/test/java/org/springframework/data/gemfire/config/xml/ContinuousQueryListenerContainerNamespaceTest.java b/src/test/java/org/springframework/data/gemfire/config/xml/ContinuousQueryListenerContainerNamespaceTest.java index f603eaf5..1788ef2c 100644 --- a/src/test/java/org/springframework/data/gemfire/config/xml/ContinuousQueryListenerContainerNamespaceTest.java +++ b/src/test/java/org/springframework/data/gemfire/config/xml/ContinuousQueryListenerContainerNamespaceTest.java @@ -50,19 +50,18 @@ import org.springframework.test.context.junit4.SpringRunner; import org.springframework.util.ErrorHandler; /** - * The ContinuousQueryListenerContainerNamespaceTest class is a test suite of test cases testing the SDG XML namespace - * for proper configuration and initialization of a ContinuousQueryListenerContainer bean component - * in the Spring context. + * Integration tests for the configuration and initialization of the SDG {@link ContinuousQueryListenerContainer} + * using the SDG XML namespace. * * @author John Blum * @see org.junit.Test * @see org.junit.runner.RunWith - * @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer - * @see org.springframework.test.context.ContextConfiguration - * @see org.springframework.test.context.junit4.SpringRunner * @see org.apache.geode.cache.client.ClientCache * @see org.apache.geode.cache.query.CqListener * @see org.apache.geode.cache.query.CqQuery + * @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer + * @see org.springframework.test.context.ContextConfiguration + * @see org.springframework.test.context.junit4.SpringRunner * @since 1.4.0 */ @RunWith(SpringRunner.class) @@ -74,6 +73,7 @@ public class ContinuousQueryListenerContainerNamespaceTest extends ClientServerI @BeforeClass public static void startGemFireServer() throws Exception { + int availablePort = findAvailablePort(); gemfireServer = run(CqCacheServerProcess.class, @@ -104,6 +104,7 @@ public class ContinuousQueryListenerContainerNamespaceTest extends ClientServerI @Test public void testContainerConfiguration() throws Exception { + assertNotNull("The ContinuousQueryListenerContainer was not properly configured!", container); assertTrue("The CQ Listener Container should be active (initialized)!", container.isActive()); assertFalse("The CQ Listener container should not be configured to auto-start!", container.isAutoStartup()); @@ -119,9 +120,10 @@ public class ContinuousQueryListenerContainerNamespaceTest extends ClientServerI assertNotNull(queries); assertEquals(3, queries.length); - List actualNames = new ArrayList(3); + List actualNames = new ArrayList<>(3); for (CqQuery query : queries) { + actualNames.add(query.getName()); assertEquals("SELECT * FROM /test-cq", query.getQueryString()); @@ -129,12 +131,9 @@ public class ContinuousQueryListenerContainerNamespaceTest extends ClientServerI CqListener cqListener = query.getCqAttributes().getCqListener(); - assertNotNull(cqListener); - - // the CqListener object should be an instance of... - // org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer.EventDispatcherAdapter - // So, get the SDG "ContinuousQueryListener"... - ContinuousQueryListener listener = TestUtils.readField("delegate", cqListener); + // The CqListener should be an instance of o.s.d.g.listener.ContinuousQueryListenerContainer.EventDispatcherAdapter + // So, get the SDG "ContinuousQueryListener" + ContinuousQueryListener listener = TestUtils.readField("listener", cqListener); assertTrue(listener instanceof ContinuousQueryListenerAdapter); assertTrue(((ContinuousQueryListenerAdapter) listener).getDelegate() instanceof GemfireMDP); diff --git a/src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTests.java b/src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTests.java index d73f2ecf..8fb00d0b 100644 --- a/src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTests.java +++ b/src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTests.java @@ -125,7 +125,7 @@ public class ContinuousQueryListenerContainerTests { assertThat(cqListenerContainer.getQueryService()).isEqualTo(mockQueryService); assertThat(cqListenerContainer.getTaskExecutor()).isInstanceOf(Executor.class); - verify(mockBeanFactory, times(1)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME)); + 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(); @@ -143,6 +143,7 @@ public class ContinuousQueryListenerContainerTests { QueryService mockQueryService = mock(QueryService.class); + when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(true); when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true); cqListenerContainer.setAutoStartup(false); @@ -160,7 +161,7 @@ public class ContinuousQueryListenerContainerTests { assertThat(cqListenerContainer.getQueryService()).isSameAs(mockQueryService); assertThat(cqListenerContainer.getTaskExecutor()).isSameAs(mockExecutor); - verify(mockBeanFactory, never()).containsBean(eq("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(poolManagerSpy, never()).find(anyString()); @@ -170,6 +171,7 @@ public class ContinuousQueryListenerContainerTests { @Test(expected = IllegalStateException.class) public void afterPropertiesSetThrowsIllegalStateExceptionWhenQueryServiceIsUninitialized() { + when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(true); when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true); try { @@ -189,7 +191,7 @@ public class ContinuousQueryListenerContainerTests { assertThat(cqListenerContainer.isAutoStartup()).isTrue(); assertThat(cqListenerContainer.isRunning()).isFalse(); - verify(mockBeanFactory, never()).containsBean(eq("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)); } @@ -240,12 +242,14 @@ public class ContinuousQueryListenerContainerTests { @Test public void eagerlyInitializePoolWithGivenPoolName() { + when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(true); when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true); cqListenerContainer.setBeanFactory(mockBeanFactory); assertThat(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)); } @@ -255,7 +259,7 @@ public class ContinuousQueryListenerContainerTests { Pool mockPool = mock(Pool.class); - when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(false); + when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(false); when(mockPool.getName()).thenReturn("TestPool"); try { @@ -268,7 +272,8 @@ public class ContinuousQueryListenerContainerTests { finally { assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue(); - verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class)); + 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(); } @@ -279,6 +284,7 @@ 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")); @@ -294,6 +300,7 @@ public class ContinuousQueryListenerContainerTests { 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(); @@ -303,7 +310,7 @@ public class ContinuousQueryListenerContainerTests { @Test(expected = IllegalArgumentException.class) public void eagerlyInitializePoolThrowsIllegalArgumentExceptionCausedByNoPoolWithGivenName() { - when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(false); + when(mockBeanFactory.containsBean(eq("TestPool"))).thenReturn(false); try { cqListenerContainer.setBeanFactory(mockBeanFactory); @@ -317,7 +324,8 @@ public class ContinuousQueryListenerContainerTests { throw expected; } finally { - verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class)); + verify(mockBeanFactory, times(1)).containsBean(eq("TestPool")); + verify(mockBeanFactory, never()).isTypeMatch(anyString(), eq(Pool.class)); verify(mockBeanFactory, never()).getBean(eq("TestPool"), eq(Pool.class)); } }