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 72c56f08..e8ecf6b6 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryDefinition.java +++ b/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryDefinition.java @@ -16,13 +16,18 @@ package org.springframework.data.gemfire.listener; +import java.util.function.Function; + +import org.apache.geode.cache.query.CqAttributes; +import org.apache.geode.cache.query.CqAttributesFactory; +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.util.Assert; /** - * Basic holder class for defining an {@link CqQuery}. Useful for configuring GemFire {@link CqQuery}s by mean of - * XML or using JavaBeans. + * Class type for defining a {@link CqQuery}. * * @author Costin Leau * @author John Blum @@ -31,25 +36,19 @@ import org.springframework.util.Assert; @SuppressWarnings("unused") public class ContinuousQueryDefinition implements InitializingBean { - private boolean durable = false; + private final boolean durable; - private ContinuousQueryListener listener; + private final ContinuousQueryListener listener; - private String name; - private String query; - - public ContinuousQueryDefinition() { - } + private final String name; + private final String query; public ContinuousQueryDefinition(String query, ContinuousQueryListener listener) { this(query, listener, false); } public ContinuousQueryDefinition(String query, ContinuousQueryListener listener, boolean durable) { - this.query = query; - this.listener = listener; - this.durable = durable; - afterPropertiesSet(); + this(null, query, listener, durable); } public ContinuousQueryDefinition(String name, String query, ContinuousQueryListener listener) { @@ -57,16 +56,13 @@ public class ContinuousQueryDefinition implements InitializingBean { } public ContinuousQueryDefinition(String name, String query, ContinuousQueryListener listener, boolean durable) { + this.name = name; this.query = query; this.listener = listener; this.durable = durable; - afterPropertiesSet(); - } - public void afterPropertiesSet() { - Assert.hasText(query, "A non-empty query is required."); - Assert.notNull(listener, "A non-null listener is required."); + afterPropertiesSet(); } /** @@ -75,34 +71,58 @@ public class ContinuousQueryDefinition implements InitializingBean { * @return a boolean indicating if the CQ is durable. */ public boolean isDurable() { - return durable; + return this.durable; } /** - * Gets the name for the CQ. + * Determines whether the CQ was named. * - * @return a String name for the CQ. + * @return a boolean value indicating whether the CQ is named. + * @see #getName() */ - public String getName() { - return name; + public boolean isNamed() { + return StringUtils.hasText(getName()); } /** - * Gets the query string that will be executed for the CQ. + * Returns a reference to the {@link ContinuousQueryListener} that will process/handle CQ event notifications. * - * @return a String value with the query to be executed for the CQ. - */ - public String getQuery() { - return query; - } - - /** - * The CQ Listener receiving events and notifications with changes from the CQ. - * - * @return the listener to be registered for the CQ. + * @return the CQ listener registered with the CQ to handle CQ events. */ public ContinuousQueryListener getListener() { - return listener; + return this.listener; } + /** + * Gets the {@link String name} of the CQ. + * + * @return the {@link String name} of the CQ. + */ + public String getName() { + return this.name; + } + + /** + * Gets the {@link String query} executed by the CQ. + * + * @return the {@link String query} executed by the CQ. + */ + public String getQuery() { + return this.query; + } + + @Override + public void afterPropertiesSet() { + Assert.hasText(query, "Query is required"); + Assert.notNull(listener, "Listener is required"); + } + + public CqAttributes toCqAttributes(Function listenerFunction) { + + CqAttributesFactory attributesFactory = new CqAttributesFactory(); + + attributesFactory.addCqListener(listenerFunction.apply(getListener())); + + return attributesFactory.create(); + } } 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 0ceee030..b795eee1 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java +++ b/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java @@ -16,11 +16,16 @@ package org.springframework.data.gemfire.listener; +import static org.springframework.data.gemfire.util.CollectionUtils.nullSafeSet; +import static org.springframework.data.gemfire.util.RuntimeExceptionFactory.newIllegalArgumentException; + import java.util.LinkedHashSet; +import java.util.Optional; import java.util.Queue; import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Executor; +import java.util.function.Supplier; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -28,8 +33,8 @@ 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.CqAttributesFactory; import org.apache.geode.cache.query.CqEvent; +import org.apache.geode.cache.query.CqException; import org.apache.geode.cache.query.CqListener; import org.apache.geode.cache.query.CqQuery; import org.apache.geode.cache.query.QueryException; @@ -49,15 +54,15 @@ import org.springframework.data.gemfire.client.support.DefaultableDelegatingPool import org.springframework.data.gemfire.client.support.DelegatingPoolAdapter; import org.springframework.data.gemfire.config.xml.GemfireConstants; import org.springframework.util.Assert; -import org.springframework.util.ClassUtils; import org.springframework.util.ErrorHandler; import org.springframework.util.StringUtils; /** - * Container providing asynchronous behaviour for GemFire Continuous Queries (CQ). + * Container providing asynchronous processing/handling for Pivotal GemFire / Apache Geode Continuous Queries (CQ). * * @author Costin Leau * @author John Blum + * @see java.util.concurrent.Executor * @see org.apache.geode.cache.RegionService * @see org.apache.geode.cache.client.Pool * @see org.apache.geode.cache.client.PoolManager @@ -76,14 +81,16 @@ import org.springframework.util.StringUtils; * @see org.springframework.core.task.TaskExecutor * @see org.springframework.data.gemfire.client.support.DefaultableDelegatingPoolAdapter * @see org.springframework.data.gemfire.client.support.DelegatingPoolAdapter + * @see org.springframework.util.ErrorHandler + * @since 1.1 */ @SuppressWarnings("unused") public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanNameAware, InitializingBean, DisposableBean, SmartLifecycle { // Default Thread name prefix is "ContinuousQueryListenerContainer-". - public static final String DEFAULT_THREAD_NAME_PREFIX = String.format("%s-", ClassUtils.getShortName( - ContinuousQueryListenerContainer.class)); + public static final String DEFAULT_THREAD_NAME_PREFIX = + String.format("%s-", ContinuousQueryListenerContainer.class.getSimpleName()); private boolean autoStartup = true; @@ -110,192 +117,167 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN private String beanName; private String poolName; + @Override public void afterPropertiesSet() { - initQueryService(eagerlyInitializePool(resolvePoolName())); + + validateQueryService(initQueryService(eagerlyInitializePool(resolvePoolName()))); initExecutor(); - initContinuousQueries(continuousQueryDefinitions); + initContinuousQueries(); + + this.initialized = true; + } + + /* (non-Javadoc) */ + private QueryService validateQueryService(QueryService queryService) { Assert.state(queryService != null, "QueryService was not properly initialized"); - initialized = true; - } - - /* (non-Javadoc) */ - String resolvePoolName() { - String poolName = this.poolName; - - if (!StringUtils.hasText(poolName)) { - String defaultPoolName = GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME; - poolName = (beanFactory != null && beanFactory.containsBean(defaultPoolName) ? defaultPoolName - : GemfireUtils.DEFAULT_POOL_NAME); - } - - return poolName; - } - - /* (non-Javadoc) */ - String eagerlyInitializePool(String poolName) { - try { - if (beanFactory != null && beanFactory.isTypeMatch(poolName, Pool.class)) { - beanFactory.getBean(poolName, Pool.class); - } - } - catch (BeansException ignore) { - Assert.notNull(PoolManager.find(poolName), - String.format("No GemFire Pool with name [%s] was found", poolName)); - } - - return poolName; - } - - /* (non-Javadoc) */ - QueryService initQueryService(String poolName) { - if (queryService == null || StringUtils.hasText(poolName)) { - queryService = DefaultableDelegatingPoolAdapter.from(DelegatingPoolAdapter.from( - PoolManager.find(poolName))).preferPool().getQueryService(queryService); - } - return queryService; } - /* (non-Javadoc) */ - Executor initExecutor() { - if (taskExecutor == null) { - taskExecutor = createDefaultTaskExecutor(); - manageExecutor = true; - } + /** + * Resolves the name of the {@link Pool} configured to handle the registered Continuous Queries. + * + * Note, the {@link Pool} must have subscription enabled. + * + * @return the {@link String name} of the {@link Pool} configured to handle the registered Continuous Queries. + */ + String resolvePoolName() { - return taskExecutor; + return Optional.ofNullable(getPoolName()) + .filter(StringUtils::hasText) + .orElseGet(() -> + Optional.ofNullable(getBeanFactory()) + .filter(it -> it.containsBean(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME)) + .map(it -> GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME) + .orElse(GemfireUtils.DEFAULT_POOL_NAME)); } /** - * Creates a default TaskExecutor. Called if no explicit TaskExecutor has been configured. - *

The default implementation builds a {@link org.springframework.core.task.SimpleAsyncTaskExecutor} - * with the specified bean name (or the class name, if no bean name is specified) as thread name prefix.

+ * Eagerly initializes the {@link Pool} with the given {@link String name}. * - * @return an instance of the TaskExecutor used to process CQ events asynchronously. - * @see org.springframework.core.task.SimpleAsyncTaskExecutor#SimpleAsyncTaskExecutor(String) + * First, this method attempts to use the configured {@link BeanFactory}, if not {@literal null}, to find a bean + * in the Spring container of type {@link Pool} having the given {@link String name }and fetch the bean, + * thereby causing the {@link Pool} bean to be initialized. + * + * 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 GemFire/Geode's {@link PoolManager}. + * + * @param poolName {@link String} containing the name of the {@link Pool} to initialize. + * @return the given {@link Pool} name. */ - protected TaskExecutor createDefaultTaskExecutor() { - return new SimpleAsyncTaskExecutor(beanName != null ? String.format("%s-", beanName) - : DEFAULT_THREAD_NAME_PREFIX); + String eagerlyInitializePool(String poolName) { + + Supplier poolNameResolver = () -> { + Assert.notNull(PoolManager.find(poolName), String.format("No Pool with name [%s] was found", poolName)); + return poolName; + }; + + return Optional.ofNullable(getBeanFactory()) + .filter(it -> it.isTypeMatch(poolName, Pool.class)) + .map(it -> { + try { + it.getBean(poolName, Pool.class); + return poolName; + } + catch (BeansException ignore) { + return poolNameResolver.get(); + } + }) + .orElseGet(poolNameResolver); } + /** + * Initializes the {@link QueryService} used to register Continuous Queries (CQ). + * + * @param poolName {@link String} containing the name of the {@link Pool} used obtain the {@link QueryService} + * if CQs are tied to a specific {@link Pool}. + * @return the initialized {@link QueryService}. + * @see org.apache.geode.cache.query.QueryService + */ + QueryService initQueryService(String poolName) { + + QueryService queryService = getQueryService(); + + if (queryService == null || StringUtils.hasText(poolName)) { + setQueryService(DefaultableDelegatingPoolAdapter.from( + DelegatingPoolAdapter.from(PoolManager.find(poolName))) + .preferPool().getQueryService(queryService)); + } + + return getQueryService(); + } + + /** + * Initialize the {@link Executor} used to process CQ events asynchronously. + * + * @return a new isntance of {@link Executor} used to process CQ events asynchronously. + * @see java.util.concurrent.Executor + */ + Executor initExecutor() { + + if (getTaskExecutor() == null) { + setTaskExecutor(createDefaultTaskExecutor()); + this.manageExecutor = true; + } + + return getTaskExecutor(); + } + + /** + * Creates a default {@link TaskExecutor}. + * + *

Called if no explicit {@link TaskExecutor} has been configured. + * + *

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 + */ + protected Executor createDefaultTaskExecutor() { + + String threadNamePrefix = Optional.ofNullable(getBeanName()) + .filter(StringUtils::hasText) + .map(it -> String.format("%s-", it)) + .orElse(DEFAULT_THREAD_NAME_PREFIX); + + return new SimpleAsyncTaskExecutor(threadNamePrefix); + } + + /** + * Initializes all the {@link CqQuery Continuous Queries} defined by the {@link ContinuousQueryDefinition defintions}. + */ + private void initContinuousQueries() { + initContinuousQueries(this.continuousQueryDefinitions); + } + + /* (non-Javadoc) */ private void initContinuousQueries(Set continuousQueryDefinitions) { - // stop the continuous query listener container if currently running... + + // Stop the ContinuousQueryListenerContainer if currently running... if (isRunning()) { stop(); } - // close any existing continuous queries... + // Close any existing continuous queries... closeQueries(); - // add current continuous queries based on the definitions from the configuration... + // Add current continuous queries based on the definitions from the configuration... for (ContinuousQueryDefinition definition : continuousQueryDefinitions) { addContinuousQuery(definition); } } - public synchronized void start() { - if (!isRunning()) { - doStart(); - running = true; - - if (logger.isDebugEnabled()) { - logger.debug("Started ContinuousQueryListenerContainer"); - } - } - } - - private void doStart() { - for (CqQuery cq : continuousQueries) { - executeQuery(cq); - } - } - - private void executeQuery(CqQuery cq) { - try { - cq.execute(); - } - catch (QueryException e) { - throw new GemfireQueryException(String.format("Could not execute query [%1$s]; state is [%2$s]", - cq.getName(), cq.getState()), e); - } - } - - @Override - public void stop(Runnable callback) { - stop(); - callback.run(); - } - - @Override - public synchronized void stop() { - if (isRunning()) { - doStop(); - running = false; - } - - if (logger.isDebugEnabled()) { - logger.debug("Stopped ContinuousQueryListenerContainer"); - } - } - - private void doStop() { - for (CqQuery cq : continuousQueries) { - try { - cq.stop(); - } - catch (Exception e) { - logger.warn(String.format("Cannot stop query [%1$s]; state is [%2$s]", cq.getName(), cq.getState()), e); - } - } - } - - @Override - public void destroy() throws Exception { - stop(); - closeQueries(); - destroyExecutor(); - initialized = false; - } - - private void closeQueries() { - for (CqQuery cq : continuousQueries) { - try { - if (!cq.isClosed()) { - cq.close(); - } - } - catch (Exception e) { - logger.warn(String.format("Cannot close query [%1$s]; state is [%2$s]", - cq.getName(), cq.getState()), e); - } - } - - continuousQueries.clear(); - } - - private void destroyExecutor() throws Exception { - if (manageExecutor) { - if (taskExecutor instanceof DisposableBean) { - ((DisposableBean) taskExecutor).destroy(); - - if (logger.isDebugEnabled()) { - logger.debug("Stopped internally-managed Task Executor."); - } - } - } - } - /** - * Determines whether this container is currently active, that is, whether it has been setup (initialized) + * Determines whether this container is currently active, i.e., whether it has been setup and initialized * but not shutdown yet. * * @return a boolean indicating whether the container is active. */ - public final boolean isActive() { - return initialized; + public boolean isActive() { + return this.initialized; } /** @@ -315,7 +297,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN */ @Override public boolean isAutoStartup() { - return autoStartup; + return this.autoStartup; } /** @@ -323,8 +305,9 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN * * @return a boolean value indicating whether the container has been started and is currently running. */ + @Override public synchronized boolean isRunning() { - return running; + return this.running; } /** @@ -338,6 +321,16 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN this.beanFactory = beanFactory; } + /** + * Returns a reference to the configured {@link BeanFactory}. + * + * @return a reference to the configured {@link BeanFactory}. + * @see org.springframework.beans.factory.BeanFactory + */ + protected BeanFactory getBeanFactory() { + return this.beanFactory; + } + /** * Set the name of the bean in the bean factory that created this bean. *

Invoked after population of normal bean properties but before an @@ -351,6 +344,15 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN this.beanName = name; } + /** + * Returns the configured {@link String bean name} of this container. + * + * @return the configured {@link String bean name} of this container. + */ + protected String getBeanName() { + return this.beanName; + } + /** * Set the underlying RegionService (GemFire Cache) used for registering Queries. * @@ -362,16 +364,41 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN } /** - * Set an ErrorHandler to be invoked in case of any uncaught exceptions thrown while processing a CQ event. - * By default there will be no ErrorHandler so that error-level logging is the only result. + * Returns a reference to all the configured/registered {@link CqQuery Continuous Queries}. * - * @param errorHandler the ErrorHandler invoked when uncaught exceptions are thrown while processing the CQ event. + * @return a reference to all the configured/registered {@link CqQuery Continuous Queries}. + * @see java.util.Queue + * @see org.apache.geode.cache.query.CqQuery + */ + protected Queue getContinuousQueries() { + return this.continuousQueries; + } + + /** + * Set an {@link ErrorHandler} to be invoked in case of any uncaught {@link Exception Exceptions} thrown + * while processing a CQ event. + * + * By default there is no {@link ErrorHandler} configured so error-level logging is the only result. + * + * @param errorHandler {@link ErrorHandler} invoked when uncaught {@link Exception Exceptions} are thrown + * while processing the CQ event. * @see org.springframework.util.ErrorHandler */ public void setErrorHandler(ErrorHandler errorHandler) { this.errorHandler = errorHandler; } + /** + * Returns an {@link Optional} reference to the configured {@link ErrorHandler} invoked when + * any unhandled {@link Exception Exceptions} are thrown when invoking CQ listeners processing CQ events. + * + * @return an {@link Optional} reference to the configured {@link ErrorHandler}. + * @see org.springframework.util.ErrorHandler + */ + protected Optional getErrorHandler() { + return Optional.ofNullable(this.errorHandler); + } + /** * Sets the phase in which this CQ listener container will start in the Spring container. * @@ -388,7 +415,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN * @see org.springframework.context.Phased#getPhase() */ public int getPhase() { - return phase; + return this.phase; } /** @@ -400,24 +427,43 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN this.poolName = poolName; } + /** + * Returns the configured {@link String pool name}. + * + * @return the configured {@link String pool name}. + */ + protected String getPoolName() { + return this.poolName; + } + /** * Attaches the given query definitions. * * @param queries set of queries */ public void setQueryListeners(Set queries) { - continuousQueryDefinitions.clear(); - continuousQueryDefinitions.addAll(queries); + this.continuousQueryDefinitions.clear(); + this.continuousQueryDefinitions.addAll(nullSafeSet(queries)); } /** * Set the GemFire QueryService used by this container to create ContinuousQueries (CQ). * - * @param service the GemFire QueryService object used by the container to create ContinuousQueries (CQ). + * @param queryService the GemFire QueryService object used by the container to create ContinuousQueries (CQ). * @see org.apache.geode.cache.query.QueryService */ - public void setQueryService(QueryService service) { - this.queryService = service; + public void setQueryService(QueryService queryService) { + this.queryService = queryService; + } + + /** + * Returns a reference to the configured {@link QueryService}. + * + * @return a reference to the configured {@link QueryService}. + * @see org.apache.geode.cache.query.QueryService + */ + protected QueryService getQueryService() { + return this.queryService; } /** @@ -434,115 +480,264 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN } /** - * Adds a Continuous Query (CQ) definition to the (potentially running) container. If the container is running, - * the listener starts receiving (matching) messages as soon as possible. + * Returns a reference to the configured {@link Executor TaskExecutor}. * - * @param definition Continuous Query (CQ) definition - * @see org.springframework.data.gemfire.listener.ContinuousQueryDefinition - * @see #doAddListener(ContinuousQueryDefinition) + * @return a reference to the configured {@link Executor TaskExecutor}. + * @see java.util.concurrent.Executor */ - public void addListener(ContinuousQueryDefinition definition) { - doAddListener(definition); - } - - private void doAddListener(ContinuousQueryDefinition definition) { - CqQuery cq = addContinuousQuery(definition); - - if (isRunning()) { - executeQuery(cq); - } - } - - private CqQuery addContinuousQuery(ContinuousQueryDefinition definition) { - try { - CqAttributesFactory continuousQueryAttributesFactory = new CqAttributesFactory(); - - continuousQueryAttributesFactory.addCqListener(new EventDispatcherAdapter(definition.getListener())); - - CqAttributes continuousQueryAttributes = continuousQueryAttributesFactory.create(); - - CqQuery cq = (StringUtils.hasText(definition.getName()) - ? queryService.newCq(definition.getName(), definition.getQuery(), continuousQueryAttributes, definition.isDurable()) - : queryService.newCq(definition.getQuery(), continuousQueryAttributes, definition.isDurable())); - - continuousQueries.add(cq); - - return cq; - } - catch (QueryException e) { - throw new GemfireQueryException("Cannot create query", e); - } - } - - private void dispatchEvent(final ContinuousQueryListener listener, final CqEvent event) { - taskExecutor.execute(() -> executeListener(listener, event)); + protected Executor getTaskExecutor() { + return this.taskExecutor; } /** - * Execute the specified listener. + * Adds a {@link ContinuousQueryDefinition Continuous Query (CQ) definition} to the (potentially running) container. * - * @param listener the ContinuousQueryListener to notify of the CQ event. - * @param event the CQ event. - * @see #handleListenerException(Throwable) + * If the container is running, the listener starts receiving (matching) messages as soon as possible. + * + * @param definition {@link ContinuousQueryDefinition Continuous Query (CQ) definition} to register. + * @see org.springframework.data.gemfire.listener.ContinuousQueryDefinition */ - protected void executeListener(ContinuousQueryListener listener, CqEvent event) { + public void addListener(ContinuousQueryDefinition definition) { + + CqQuery query = addContinuousQuery(definition); + + if (isRunning()) { + execute(query); + } + } + + /* (non-Javadoc) */ + CqQuery addContinuousQuery(ContinuousQueryDefinition definition) { + + try { + CqAttributes attributes = definition.toCqAttributes(this::newCqListener); + + CqQuery query = (definition.isNamed() ? newNamedContinuousQuery(definition, attributes) + : newUnnamedContinuousQuery(definition, attributes)); + + return manage(query); + } + catch (QueryException cause) { + throw new GemfireQueryException(String.format("Unable to create query [%s]", definition.getQuery()), cause); + } + } + + /* (non-Javadoc) */ + protected CqListener newCqListener(ContinuousQueryListener listener) { + return new EventDispatcherAdapter(listener); + } + + /* (non-Javadoc) */ + private CqQuery newNamedContinuousQuery(ContinuousQueryDefinition definition, CqAttributes attributes) + throws QueryException { + + return getQueryService().newCq(definition.getName(), definition.getQuery(), attributes, definition.isDurable()); + } + + /* (non-Javadoc) */ + private CqQuery newUnnamedContinuousQuery(ContinuousQueryDefinition definition, CqAttributes attributes) + throws CqException { + + return getQueryService().newCq(definition.getQuery(), attributes, definition.isDurable()); + } + + /* (non-Javadoc) */ + private CqQuery manage(CqQuery query) { + getContinuousQueries().add(query); + return query; + } + + @Override + public synchronized void start() { + + if (!isRunning()) { + doStart(); + this.running = true; + + if (logger.isDebugEnabled()) { + logger.debug("Started ContinuousQueryListenerContainer"); + } + } + } + + /* (non-Javadoc) */ + void doStart() { + getContinuousQueries().forEach(this::execute); + } + + /* (non-Javadoc) */ + private void execute(CqQuery query) { + try { + query.execute(); + } + catch (QueryException cause) { + throw new GemfireQueryException(String.format("Could not execute query [%1$s]; state is [%2$s]", + query.getName(), query.getState()), cause); + } + } + + /** + * Asynchronously dispatches the {@link CqEvent CQ event} to the targeted {@link ContinuousQueryListener}. + * + * @param listener {@link ContinuousQueryListener} which will process/handle the {@link CqEvent CQ event}. + * @param event {@link CqEvent CQ event} to process. + * @see org.springframework.data.gemfire.listener.ContinuousQueryListener + * @see org.apache.geode.cache.query.CqEvent + */ + protected void dispatchEvent(ContinuousQueryListener listener, CqEvent event) { + getTaskExecutor().execute(() -> notify(listener, event)); + } + + /** + * Invoke the specified {@link ContinuousQueryListener listener} to process/handle the {@link CqEvent CQ event}. + * + * @param listener {@link ContinuousQueryListener} to notify of the {@link CqEvent CQ event}. + * @param event {@link CqEvent CQ event} to process/handle. + * @see #handleListenerError(Throwable) + */ + private void notify(ContinuousQueryListener listener, CqEvent event) { try { listener.onEvent(event); } - catch (Throwable ex) { - handleListenerException(ex); + catch (Throwable cause) { + handleListenerError(cause); } } /** - * Handle the given exception that arose during listener execution. - *

The default implementation logs the exception at error level. - * This can be overridden in subclasses. + * Invokes the configured {@link ErrorHandler} (if any) to handle the {@link Exception} thrown by the CQ listener. * - * @param e the exception to handle + * Logs at warning level if no {@link ErrorHandler} was configured. + * + * Logs at debug level if the CQ listener container was shutdown at the time when the CQ listener + * {@link Exception} was thrown. + * + * @param cause {@link Throwable uncaught error} thrown during normal CQ event processing. + * @see #setErrorHandler(ErrorHandler) + * @see #getErrorHandler() */ - protected void handleListenerException(Throwable e) { - if (isActive()) { - // Regular case: failed while active. - // Invoke ErrorHandler if available. - invokeErrorHandler(e); - } - else { - // Rare case: listener thread failed after container shutdown. - // Log at debug level, to avoid spamming the shutdown logger. - logger.debug("Listener exception after container shutdown", e); + private void handleListenerError(Throwable cause) { + + getErrorHandler() + .filter(errorHandler -> { + + boolean active = this.isActive(); + + if (!active && logger.isDebugEnabled()) { + 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); } } - /** - * Invoke the registered ErrorHandler, if any. Log at error level otherwise. - * - * @param e the uncaught error that arose during event processing. - * @see #setErrorHandler - */ - protected void invokeErrorHandler(Throwable e) { - if (this.errorHandler != null) { - this.errorHandler.handleError(e); + @Override + @SuppressWarnings("all") + public void stop(Runnable callback) { + stop(); + callback.run(); + } + + @Override + public synchronized void stop() { + + if (isRunning()) { + doStop(); + this.running = false; } - else if (logger.isWarnEnabled()) { - logger.warn("Execution of the CQ event listener failed, and no ErrorHandler has been set.", e); + + if (logger.isDebugEnabled()) { + logger.debug("Stopped ContinuousQueryListenerContainer"); } } - private class EventDispatcherAdapter implements CqListener { + /* (non-Javadoc) */ + void doStop() { - private final ContinuousQueryListener delegate; + 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]", + query.getName(), query.getState()), cause); + } + } + }); + } - private EventDispatcherAdapter(ContinuousQueryListener delegate) { - this.delegate = delegate; + @Override + public void destroy() throws Exception { + stop(); + closeQueries(); + destroyExecutor(); + this.initialized = false; + } + + /* (non-Javadoc) */ + 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().clear(); + } + + /* (non-Javadoc) */ + private void destroyExecutor() { + + Optional.ofNullable(getTaskExecutor()) + .filter(it -> this.manageExecutor) + .filter(it -> it instanceof DisposableBean) + .ifPresent(it -> { + try { + ((DisposableBean) it).destroy(); + + if (logger.isDebugEnabled()) { + logger.debug(String.format("Stopped internally-managed TaskExecutor [%s]", it)); + } + } + catch (Exception ignore) { + logger.warn(String.format("Failed to properly destroy the managed TaskExecutor [%s]", it)); + } + }); + } + + protected class EventDispatcherAdapter implements CqListener { + + private final ContinuousQueryListener listener; + + protected EventDispatcherAdapter(ContinuousQueryListener listener) { + this.listener = Optional.ofNullable(listener) + .orElseThrow(() -> newIllegalArgumentException("ContinuousQueryListener is required")); + } + + protected ContinuousQueryListener getListener() { + return this.listener; } public void onError(CqEvent event) { - dispatchEvent(delegate, event); + dispatchEvent(getListener(), event); } public void onEvent(CqEvent event) { - dispatchEvent(delegate, event); + dispatchEvent(getListener(), event); } public void close() { 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 cb961d5e..d73f2ecf 100644 --- a/src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTests.java +++ b/src/test/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainerTests.java @@ -17,14 +17,16 @@ package org.springframework.data.gemfire.listener; -import static org.hamcrest.Matchers.equalTo; -import static org.hamcrest.Matchers.instanceOf; -import static org.hamcrest.Matchers.is; -import static org.hamcrest.Matchers.nullValue; -import static org.hamcrest.Matchers.sameInstance; -import static org.junit.Assert.assertThat; +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.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; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; @@ -33,21 +35,32 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyZeroInteractions; import static org.mockito.Mockito.when; +import java.util.Arrays; +import java.util.List; import java.util.concurrent.Executor; import org.apache.geode.cache.RegionService; import org.apache.geode.cache.client.Pool; +import org.apache.geode.cache.query.CqAttributes; +import org.apache.geode.cache.query.CqEvent; +import org.apache.geode.cache.query.CqException; +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.Rule; import org.junit.Test; -import org.junit.rules.ExpectedException; +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.TestUtils; import org.springframework.data.gemfire.config.xml.GemfireConstants; +import org.springframework.util.ErrorHandler; /** * Unit tests for {@link ContinuousQueryListenerContainer}. @@ -60,43 +73,57 @@ import org.springframework.data.gemfire.config.xml.GemfireConstants; * @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer * @since 1.8.0 */ +@RunWith(MockitoJUnitRunner.class) public class ContinuousQueryListenerContainerTests { - @Rule - public ExpectedException exception = ExpectedException.none(); + @Mock + private BeanFactory mockBeanFactory; - private ContinuousQueryListenerContainer listenerContainer; + private ContinuousQueryListenerContainer cqListenerContainer; + private CqQuery mockCqQuery(String name, String query, CqAttributes attributes, boolean durable) { + + CqQuery mockQuery = mock(CqQuery.class); + + when(mockQuery.getName()).thenReturn(name); + when(mockQuery.getQueryString()).thenReturn(query); + when(mockQuery.getCqAttributes()).thenReturn(attributes); + when(mockQuery.isDurable()).thenReturn(durable); + + return mockQuery; + } @Before public void setup() { - listenerContainer = new ContinuousQueryListenerContainer(); + cqListenerContainer = spy(new ContinuousQueryListenerContainer()); } @Test public void afterPropertiesSetIsAutoStart() throws Exception { - BeanFactory mockBeanFactory = mock(BeanFactory.class); + Pool mockPool = mock(Pool.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(mockPool.getQueryService()).thenReturn(mockQueryService); try { PoolManagerImpl.getPMI().register(mockPool); - listenerContainer.setAutoStartup(true); - listenerContainer.setBeanFactory(mockBeanFactory); - listenerContainer.afterPropertiesSet(); + cqListenerContainer.setAutoStartup(true); + cqListenerContainer.setBeanFactory(mockBeanFactory); + cqListenerContainer.afterPropertiesSet(); } finally { - assertThat(PoolManagerImpl.getPMI().unregister(mockPool), is(true)); - assertThat(listenerContainer.isActive(), is(true)); - assertThat(listenerContainer.isAutoStartup(), is(true)); - assertThat(listenerContainer.isRunning(), is(false)); - assertThat(TestUtils.readField("queryService", listenerContainer), is(equalTo(mockQueryService))); - assertThat(TestUtils.readField("taskExecutor", listenerContainer), is(instanceOf(Executor.class))); + 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(1)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME)); verify(mockBeanFactory, times(1)).isTypeMatch(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME), eq(Pool.class)); @@ -109,22 +136,29 @@ public class ContinuousQueryListenerContainerTests { @Test public void afterPropertiesSetIsManualStart() { - BeanFactory mockBeanFactory = mock(BeanFactory.class); - QueryService mockQueryService = mock(QueryService.class); + + Executor mockExecutor = mock(Executor.class); PoolManagerImpl poolManagerSpy = spy(PoolManagerImpl.getPMI()); + QueryService mockQueryService = mock(QueryService.class); + when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true); - listenerContainer.setAutoStartup(false); - listenerContainer.setBeanFactory(mockBeanFactory); - listenerContainer.setPoolName("TestPool"); - listenerContainer.setQueryService(mockQueryService); - listenerContainer.afterPropertiesSet(); + cqListenerContainer.setAutoStartup(false); + cqListenerContainer.setBeanFactory(mockBeanFactory); + cqListenerContainer.setPoolName("TestPool"); + cqListenerContainer.setQueryService(mockQueryService); + cqListenerContainer.setTaskExecutor(mockExecutor); + cqListenerContainer.afterPropertiesSet(); - assertThat(listenerContainer.isActive(), is(true)); - assertThat(listenerContainer.isAutoStartup(), is(false)); - assertThat(listenerContainer.isRunning(), is(false)); + 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); verify(mockBeanFactory, never()).containsBean(eq("TestPool")); verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class)); @@ -133,146 +167,183 @@ public class ContinuousQueryListenerContainerTests { verifyZeroInteractions(mockQueryService); } - @Test - public void afterPropertiesSetThrowsIllegalStateExceptionWhenQueryServicesIsUninitialized() { - exception.expect(IllegalStateException.class); - exception.expectCause(is(nullValue(Throwable.class))); - exception.expectMessage("QueryService was not properly initialized"); - - try { - listenerContainer.setPoolName("TestPool"); - listenerContainer.afterPropertiesSet(); - } - finally { - assertThat(listenerContainer.isActive(), is(false)); - assertThat(listenerContainer.isAutoStartup(), is(true)); - assertThat(listenerContainer.isRunning(), is(false)); - } - } - - @Test - public void resolvesToProvidedPoolName() { - listenerContainer.setPoolName("TestPool"); - assertThat(listenerContainer.resolvePoolName(), is(equalTo("TestPool"))); - } - - @Test - public void resolvesToGemFireDefaultPoolName() { - listenerContainer.setPoolName(null); - assertThat(listenerContainer.resolvePoolName(), is(equalTo(GemfireUtils.DEFAULT_POOL_NAME))); - } - - @Test - public void resolvesToGemFireDefaultPoolNameWhenBeanFactoryDoesNotContainNamedPool() { - BeanFactory mockBeanFactory = mock(BeanFactory.class); - - when(mockBeanFactory.containsBean(anyString())).thenReturn(false); - - listenerContainer.setBeanFactory(mockBeanFactory); - listenerContainer.setPoolName(null); - - assertThat(listenerContainer.resolvePoolName(), is(equalTo(GemfireUtils.DEFAULT_POOL_NAME))); - - verify(mockBeanFactory, times(1)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME)); - } - - @Test - public void resolvesToSpringDataGemFireDefaultPoolName() { - BeanFactory mockBeanFactory = mock(BeanFactory.class); - - when(mockBeanFactory.containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME))).thenReturn(true); - - listenerContainer.setBeanFactory(mockBeanFactory); - listenerContainer.setPoolName(null); - - assertThat(listenerContainer.resolvePoolName(), is(equalTo(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME))); - - verify(mockBeanFactory, times(1)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME)); - } - - @Test - public void eagerlyInitializesNamedPool() { - BeanFactory mockBeanFactory = mock(BeanFactory.class); + @Test(expected = IllegalStateException.class) + public void afterPropertiesSetThrowsIllegalStateExceptionWhenQueryServiceIsUninitialized() { when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true); - listenerContainer.setBeanFactory(mockBeanFactory); - - assertThat(listenerContainer.eagerlyInitializePool("TestPool"), is(equalTo("TestPool"))); - - verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class)); - verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class)); - } - - @Test - public void eagerlyInitializePoolFindsRegisteredPoolByNameInGemFire() { - BeanFactory mockBeanFactory = mock(BeanFactory.class); - Pool mockPool = mock(Pool.class); - - 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"); - - listenerContainer.setBeanFactory(mockBeanFactory); - try { - PoolManagerImpl.getPMI().register(mockPool); - assertThat(listenerContainer.eagerlyInitializePool("TestPool"), is(equalTo("TestPool"))); + cqListenerContainer.setBeanFactory(mockBeanFactory); + cqListenerContainer.setPoolName("TestPool"); + cqListenerContainer.afterPropertiesSet(); + } + catch (IllegalStateException expected) { + + assertThat(expected).hasMessage("QueryService was not properly initialized"); + assertThat(expected).hasNoCause(); + + throw expected; } finally { - assertThat(PoolManagerImpl.getPMI().unregister(mockPool), is(true)); - verify(mockPool, times(2)).getName(); - } - } + assertThat(cqListenerContainer.isActive()).isFalse(); + assertThat(cqListenerContainer.isAutoStartup()).isTrue(); + assertThat(cqListenerContainer.isRunning()).isFalse(); - @Test - public void eagerlyInitializePoolThrowsIllegalArgumentExceptionCausedByNoSuchBeanDefinitionException() { - BeanFactory mockBeanFactory = mock(BeanFactory.class); - - when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true); - when(mockBeanFactory.getBean(eq("TestPool"), eq(Pool.class))).thenThrow( - new NoSuchBeanDefinitionException("test")); - - try { - exception.expect(IllegalArgumentException.class); - exception.expectCause(is(nullValue(Throwable.class))); - exception.expectMessage("No GemFire Pool with name [TestPool] was found"); - - listenerContainer.setBeanFactory(mockBeanFactory); - listenerContainer.eagerlyInitializePool("TestPool"); - } - finally { + verify(mockBeanFactory, never()).containsBean(eq("TestPool")); verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class)); verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class)); } } @Test - public void initQueryServiceReturnsProvidedQueryService() { + public void resolvePoolNameReturnsConfiguredPoolName() { + + cqListenerContainer.setPoolName("TestPool"); + + assertThat(cqListenerContainer.resolvePoolName()).isEqualTo("TestPool"); + } + + @Test + public void resolvePoolNameReturnsGemFireDefaultPoolName() { + + cqListenerContainer.setPoolName(null); + + assertThat(cqListenerContainer.resolvePoolName()).isEqualTo(GemfireUtils.DEFAULT_POOL_NAME); + } + + @Test + public void resolvePoolNameReturnsGemFireDefaultPoolNameWhenBeanFactoryDoesNotContainNamedPool() { + + when(mockBeanFactory.containsBean(anyString())).thenReturn(false); + + cqListenerContainer.setBeanFactory(mockBeanFactory); + cqListenerContainer.setPoolName(""); + + assertThat(cqListenerContainer.resolvePoolName()).isEqualTo(GemfireUtils.DEFAULT_POOL_NAME); + + verify(mockBeanFactory, times(1)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME)); + } + + @Test + public void resolvePoolNameReturnsSpringDataGemFireDefaultPoolName() { + + when(mockBeanFactory.containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME))).thenReturn(true); + + cqListenerContainer.setBeanFactory(mockBeanFactory); + cqListenerContainer.setPoolName(" "); + + assertThat(cqListenerContainer.resolvePoolName()).isEqualTo(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME); + + verify(mockBeanFactory, times(1)).containsBean(eq(GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME)); + } + + @Test + public void eagerlyInitializePoolWithGivenPoolName() { + + when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(true); + + cqListenerContainer.setBeanFactory(mockBeanFactory); + + assertThat(cqListenerContainer.eagerlyInitializePool("TestPool")).isEqualTo("TestPool"); + + verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class)); + verify(mockBeanFactory, times(1)).getBean(eq("TestPool"), eq(Pool.class)); + } + + @Test + public void eagerlyInitializePoolFindsRegisteredPoolWithTheGivenName() { + + Pool mockPool = mock(Pool.class); + + when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(false); + when(mockPool.getName()).thenReturn("TestPool"); + + try { + PoolManagerImpl.getPMI().register(mockPool); + + cqListenerContainer.setBeanFactory(mockBeanFactory); + + assertThat(cqListenerContainer.eagerlyInitializePool("TestPool")).isEqualTo("TestPool"); + } + finally { + assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue(); + + verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class)); + verify(mockBeanFactory, never()).getBean(eq("TestPool"), eq(Pool.class)); + verify(mockPool, atLeast(1)).getName(); + } + } + + @Test + public void eagerlyInitializePoolFindsRegisteredPoolWithTheGivenNameWhenBeanFactoryGetBeanThrowsBeansException() { + + Pool mockPool = mock(Pool.class); + + 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"); + + cqListenerContainer.setBeanFactory(mockBeanFactory); + + try { + PoolManagerImpl.getPMI().register(mockPool); + + assertThat(cqListenerContainer.eagerlyInitializePool("TestPool")).isEqualTo("TestPool"); + } + finally { + assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue(); + + 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(); + } + } + + @Test(expected = IllegalArgumentException.class) + public void eagerlyInitializePoolThrowsIllegalArgumentExceptionCausedByNoPoolWithGivenName() { + + when(mockBeanFactory.isTypeMatch(eq("TestPool"), eq(Pool.class))).thenReturn(false); + + try { + cqListenerContainer.setBeanFactory(mockBeanFactory); + cqListenerContainer.eagerlyInitializePool("TestPool"); + } + catch (IllegalArgumentException expected) { + + assertThat(expected).hasMessage("No Pool with name [TestPool] was found"); + assertThat(expected).hasNoCause(); + + throw expected; + } + finally { + verify(mockBeanFactory, times(1)).isTypeMatch(eq("TestPool"), eq(Pool.class)); + verify(mockBeanFactory, never()).getBean(eq("TestPool"), eq(Pool.class)); + } + } + + @Test + public void initQueryServiceReturnsConfiguredQueryService() { + QueryService mockQueryService = mock(QueryService.class); - listenerContainer.setQueryService(mockQueryService); + cqListenerContainer.setQueryService(mockQueryService); - assertThat(listenerContainer.initQueryService(null), is(sameInstance(mockQueryService))); + assertThat(cqListenerContainer.initQueryService("TestPool")).isSameAs(mockQueryService); verifyZeroInteractions(mockQueryService); } @Test - public void initializesQueryServiceFromContainer() { - QueryService mockQueryService = mock(QueryService.class); - - listenerContainer.setQueryService(mockQueryService); - - assertThat(listenerContainer.initQueryService("TestPool"), is(equalTo(mockQueryService))); - - verifyZeroInteractions(mockQueryService); + public void initQueryServiceReturnsNull() { + assertThat(cqListenerContainer.initQueryService(null)).isNull(); } @Test public void initializesQueryServiceFromPool() { + Pool mockPool = mock(Pool.class); + QueryService mockQueryService = mock(QueryService.class); when(mockPool.getName()).thenReturn("TestPool"); @@ -280,19 +351,22 @@ public class ContinuousQueryListenerContainerTests { try { PoolManagerImpl.getPMI().register(mockPool); - listenerContainer.setQueryService(null); - assertThat(listenerContainer.initQueryService("TestPool"), is(equalTo(mockQueryService))); + + assertThat(cqListenerContainer.getQueryService()).isNull(); + assertThat(cqListenerContainer.initQueryService("TestPool")).isEqualTo(mockQueryService); } finally { - assertThat(PoolManagerImpl.getPMI().unregister(mockPool), is(true)); - verify(mockPool, times(2)).getName(); + assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue(); + + verify(mockPool, atLeast(1)).getName(); verify(mockPool, times(1)).getQueryService(); verifyZeroInteractions(mockQueryService); } } @Test - public void initializesQueryServiceFromPoolInThePresenceOfAProvidedQueryService() { + public void initializesQueryServiceFromPoolIgnoresConfiguredQueryService() { + Pool mockPool = mock(Pool.class); QueryService mockQueryServiceOne = mock(QueryService.class); @@ -303,12 +377,16 @@ public class ContinuousQueryListenerContainerTests { try { PoolManagerImpl.getPMI().register(mockPool); - listenerContainer.setQueryService(mockQueryServiceTwo); - assertThat(listenerContainer.initQueryService("TestPool"), is(equalTo(mockQueryServiceOne))); + + cqListenerContainer.setQueryService(mockQueryServiceTwo); + + assertThat(cqListenerContainer.getQueryService()).isSameAs(mockQueryServiceTwo); + assertThat(cqListenerContainer.initQueryService("TestPool")).isEqualTo(mockQueryServiceOne); } finally { - assertThat(PoolManagerImpl.getPMI().unregister(mockPool), is(true)); - verify(mockPool, times(2)).getName(); + assertThat(PoolManagerImpl.getPMI().unregister(mockPool)).isTrue(); + + verify(mockPool, atLeast(1)).getName(); verify(mockPool, times(1)).getQueryService(); verifyZeroInteractions(mockQueryServiceOne); verifyZeroInteractions(mockQueryServiceTwo); @@ -317,45 +395,374 @@ public class ContinuousQueryListenerContainerTests { @Test public void initExecutorReturnsProvidedExecutor() { + Executor mockExecutor = mock(Executor.class); - listenerContainer.setTaskExecutor(mockExecutor); + cqListenerContainer.setTaskExecutor(mockExecutor); - assertThat(listenerContainer.initExecutor(), is(sameInstance(mockExecutor))); + assertThat(cqListenerContainer.initExecutor()).isSameAs(mockExecutor); verifyZeroInteractions(mockExecutor); } @Test public void initializesDefaultTaskExecutor() { - assertThat(listenerContainer.initExecutor(), is(instanceOf(Executor.class))); + assertThat(cqListenerContainer.getTaskExecutor()).isNull(); + assertThat(cqListenerContainer.initExecutor()).isInstanceOf(Executor.class); + } + + @Test + public void setAndGetAutoStartup() { + + assertThat(cqListenerContainer.isAutoStartup()).isTrue(); + + cqListenerContainer.setAutoStartup(false); + + assertThat(cqListenerContainer.isAutoStartup()).isFalse(); + + cqListenerContainer.setAutoStartup(true); + + assertThat(cqListenerContainer.isAutoStartup()).isTrue(); } @Test public void setCacheSetsQueryService() { + QueryService mockQueryService = mock(QueryService.class); + RegionService mockRegionService = mock(RegionService.class); when(mockRegionService.getQueryService()).thenReturn(mockQueryService); - listenerContainer.setCache(mockRegionService); + cqListenerContainer.setCache(mockRegionService); - assertThat(listenerContainer.initQueryService(null), is(equalTo(mockQueryService))); + assertThat(cqListenerContainer.initQueryService(null)).isEqualTo(mockQueryService); verify(mockRegionService, times(1)).getQueryService(); verifyZeroInteractions(mockQueryService); } @Test - public void setAndGetAutoStartup() { - assertThat(listenerContainer.isAutoStartup(), is(true)); + public void addListenerCreatesCqQueryAndExecutesQueryWhenRunning() throws Exception { - listenerContainer.setAutoStartup(false); + ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class); - assertThat(listenerContainer.isAutoStartup(), is(false)); + ContinuousQueryDefinition definition = + new ContinuousQueryDefinition("TestQuery", "SELECT * FROM /Utilization u WHERE u.value > 100", mockListener); - listenerContainer.setAutoStartup(true); + CqQuery mockQuery = mock(CqQuery.class); - assertThat(listenerContainer.isAutoStartup(), is(true)); + doReturn(mockQuery).when(cqListenerContainer).addContinuousQuery(eq(definition)); + when(cqListenerContainer.isRunning()).thenReturn(true); + + cqListenerContainer.addListener(definition); + + verify(cqListenerContainer, times(1)).addContinuousQuery(eq(definition)); + verify(cqListenerContainer, times(1)).isRunning(); + verify(mockQuery, times(1)).execute(); } + + @Test(expected = GemfireQueryException.class) + public void addContinuousQueryThrowsQueryException() throws Exception { + + QueryService mockQueryService = mock(QueryService.class); + + when(mockQueryService.newCq(anyString(), any(CqAttributes.class), anyBoolean())) + .thenThrow(new CqException("TEST")); + + ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class); + + ContinuousQueryDefinition definition = + new ContinuousQueryDefinition("SELECT * FROM /Utilization u WHERE u.value > 100", + mockListener, false); + + cqListenerContainer.setQueryService(mockQueryService); + + try { + cqListenerContainer.addContinuousQuery(definition); + } + catch (GemfireQueryException expected) { + + assertThat(expected).hasMessageStartingWith("Unable to create query [SELECT * FROM /Utilization u WHERE u.value > 100]"); + assertThat(expected).hasCauseInstanceOf(QueryException.class); + assertThat(expected.getCause()).hasMessage("TEST"); + assertThat(expected.getCause()).hasNoCause(); + + throw expected; + + } + finally { + assertThat(cqListenerContainer.getContinuousQueries()).isEmpty(); + } + } + + @Test + public void addManagedNamedContinuousQuery() throws Exception { + + QueryService mockQueryService = mock(QueryService.class); + + when(mockQueryService.newCq(anyString(), anyString(), any(CqAttributes.class), anyBoolean())) + .thenAnswer(invocation -> mockCqQuery(invocation.getArgument(0), invocation.getArgument(1), + invocation.getArgument(2), invocation.getArgument(3))); + + ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class); + + ContinuousQueryDefinition definition = + new ContinuousQueryDefinition("TestQuery", "SELECT * FROM /Utilization u WHERE u.value > 100", + mockListener, true); + + cqListenerContainer.setQueryService(mockQueryService); + + CqQuery query = cqListenerContainer.addContinuousQuery(definition); + + assertThat(query).isNotNull(); + assertThat(query.isDurable()).isTrue(); + assertThat(query.getName()).isEqualTo("TestQuery"); + assertThat(query.getQueryString()).isEqualTo("SELECT * FROM /Utilization u WHERE u.value > 100"); + assertThat(query.isRunning()).isFalse(); + + CqAttributes attributes = query.getCqAttributes(); + + assertThat(attributes).isNotNull(); + assertThat(attributes.getCqListener()).isInstanceOf(ContinuousQueryListenerContainer.EventDispatcherAdapter.class); + + ContinuousQueryListenerContainer.EventDispatcherAdapter eventDispatcherAdapter = + (ContinuousQueryListenerContainer.EventDispatcherAdapter) attributes.getCqListener(); + + assertThat(eventDispatcherAdapter.getListener()).isEqualTo(mockListener); + assertThat(cqListenerContainer.getContinuousQueries().peek()).isEqualTo(query); + } + + @Test + public void addManagedUnnamedContinuousQuery() throws Exception { + + QueryService mockQueryService = mock(QueryService.class); + + when(mockQueryService.newCq(anyString(), any(CqAttributes.class), anyBoolean())) + .thenAnswer(invocation -> mockCqQuery(null, invocation.getArgument(0), invocation.getArgument(1), + invocation.getArgument(2))); + + ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class); + + ContinuousQueryDefinition definition = + new ContinuousQueryDefinition("SELECT * FROM /Utilization u WHERE u.value > 100", + mockListener, false); + + cqListenerContainer.setQueryService(mockQueryService); + + CqQuery query = cqListenerContainer.addContinuousQuery(definition); + + assertThat(query).isNotNull(); + assertThat(query.isDurable()).isFalse(); + assertThat(query.getName()).isNull(); + assertThat(query.getQueryString()).isEqualTo("SELECT * FROM /Utilization u WHERE u.value > 100"); + assertThat(query.isRunning()).isFalse(); + + CqAttributes attributes = query.getCqAttributes(); + + assertThat(attributes).isNotNull(); + assertThat(attributes.getCqListener()).isInstanceOf(ContinuousQueryListenerContainer.EventDispatcherAdapter.class); + + ContinuousQueryListenerContainer.EventDispatcherAdapter eventDispatcherAdapter = + (ContinuousQueryListenerContainer.EventDispatcherAdapter) attributes.getCqListener(); + + assertThat(eventDispatcherAdapter.getListener()).isEqualTo(mockListener); + assertThat(cqListenerContainer.getContinuousQueries().peek()).isEqualTo(query); + } + + @Test + public void cqListenerContainerStartsWhenNotRunning() throws Exception { + + CqQuery mockQueryOne = mock(CqQuery.class); + CqQuery mockQueryTwo = mock(CqQuery.class); + + cqListenerContainer.getContinuousQueries().add(mockQueryOne); + cqListenerContainer.getContinuousQueries().add(mockQueryTwo); + + assertThat(cqListenerContainer.isRunning()).isFalse(); + + cqListenerContainer.start(); + + assertThat(cqListenerContainer.isRunning()).isTrue(); + + verify(mockQueryOne, times(1)).execute(); + verify(mockQueryTwo, times(1)).execute(); + } + + @Test(expected = GemfireQueryException.class) + public void cqListenerContainerStartHandlesCqException() throws Exception { + + CqQuery mockQueryOne = mock(CqQuery.class); + CqQuery mockQueryTwo = mock(CqQuery.class); + + CqState mockQueryState = mock(CqState.class); + + cqListenerContainer.getContinuousQueries().add(mockQueryOne); + cqListenerContainer.getContinuousQueries().add(mockQueryTwo); + + when(mockQueryOne.getName()).thenReturn("ONE"); + when(mockQueryOne.getState()).thenReturn(mockQueryState); + when(mockQueryState.toString()).thenReturn("FAILED"); + doThrow(new CqException("ONE")).when(mockQueryOne).execute(); + + try { + cqListenerContainer.start(); + } + catch (GemfireQueryException cause) { + + assertThat(cause).hasMessageStartingWith("Could not execute query [ONE]; state is [FAILED]"); + assertThat(cause).hasCauseInstanceOf(CqException.class); + assertThat(cause.getCause()).hasMessage("ONE"); + assertThat(cause.getCause()).hasNoCause(); + + throw cause; + } + finally { + assertThat(cqListenerContainer.isRunning()).isFalse(); + + verify(mockQueryOne, times(1)).execute(); + verify(mockQueryOne, times(1)).getName(); + verify(mockQueryOne, times(1)).getState(); + verify(mockQueryTwo, never()).execute(); + } + } + + @Test + public void cqListenerContainerDoesNotStartWhenAlreadyRunning() { + + when(cqListenerContainer.isRunning()).thenReturn(true); + + cqListenerContainer.start(); + + verify(cqListenerContainer, never()).doStart(); + } + + @Test + public void dispatchEventNotifiesListenerOfCqEvent() { + + Executor mockExecutor = mock(Executor.class); + + doAnswer(invocation -> { + invocation.getArgument(0).run(); + return null; + }).when(mockExecutor).execute(any()); + + ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class); + + CqEvent mockEvent = mock(CqEvent.class); + + cqListenerContainer.setTaskExecutor(mockExecutor); + cqListenerContainer.dispatchEvent(mockListener, mockEvent); + + verify(mockExecutor, times(1)).execute(isA(Runnable.class)); + verify(mockListener, times(1)).onEvent(eq(mockEvent)); + } + + @Test + public void dispatchEventInvokesConfiguredErrorHandlerOnListenerException() { + + RuntimeException expectedCause = new RuntimeException("TEST"); + + Executor mockExecutor = mock(Executor.class); + + doAnswer(invocation -> { + invocation.getArgument(0).run(); + return null; + }).when(mockExecutor).execute(any()); + + ErrorHandler mockErrorHandler = mock(ErrorHandler.class); + + ContinuousQueryListener mockListener = mock(ContinuousQueryListener.class); + + doThrow(expectedCause).when(mockListener).onEvent(any()); + + CqEvent mockEvent = mock(CqEvent.class); + + cqListenerContainer.setErrorHandler(mockErrorHandler); + cqListenerContainer.setTaskExecutor(mockExecutor); + + doReturn(true).when(cqListenerContainer).isActive(); + + cqListenerContainer.dispatchEvent(mockListener, mockEvent); + + verify(mockExecutor, times(1)).execute(isA(Runnable.class)); + verify(mockListener, times(1)).onEvent(eq(mockEvent)); + verify(mockErrorHandler, times(1)).handleError(eq(expectedCause)); + } + + @Test + public void stopStopsCqsCallsRunnableHandlesExceptionsOnCqQueryStopWhenRunning() throws Exception { + + CqQuery mockQueryOne = mock(CqQuery.class); + CqQuery mockQueryTwo = mock(CqQuery.class); + CqQuery mockQueryThree = mock(CqQuery.class); + + doThrow(new CqException("TWO")).when(mockQueryTwo).stop(); + + Runnable mockRunnable = mock(Runnable.class); + + cqListenerContainer.getContinuousQueries().add(mockQueryOne); + cqListenerContainer.getContinuousQueries().add(mockQueryTwo); + cqListenerContainer.getContinuousQueries().add(mockQueryThree); + + doReturn(true).when(cqListenerContainer).isRunning(); + + cqListenerContainer.stop(mockRunnable); + + verify(cqListenerContainer, times(1)).stop(); + verify(mockQueryOne, times(1)).stop(); + verify(mockQueryTwo, times(1)).stop(); + verify(mockQueryThree, times(1)).stop(); + verify(mockRunnable, times(1)).run(); + } + + @Test + public void stopDoesNothingWhenContainerIsNotRunning() { + + when(cqListenerContainer.isRunning()).thenReturn(false); + + cqListenerContainer.stop(); + + verify(cqListenerContainer, never()).doStop(); + } + + @Test + public void destroyIsSuccessful() throws Exception { + + CqQuery mockQueryOne = mock(CqQuery.class); + CqQuery mockQueryTwo = mock(CqQuery.class); + CqQuery mockQueryThree = mock(CqQuery.class); + CqQuery mockQueryFour = mock(CqQuery.class); + + when(mockQueryTwo.isClosed()).thenReturn(true); + doThrow(new RuntimeException("THREE")).when(mockQueryThree).close(); + + List queries = Arrays.asList(mockQueryOne, mockQueryTwo, mockQueryThree, mockQueryFour); + + cqListenerContainer.getContinuousQueries().addAll(queries); + + DisposableExecutorBean mockDisposableExecutorBean = mock(DisposableExecutorBean.class); + + doReturn(mockDisposableExecutorBean).when(cqListenerContainer).createDefaultTaskExecutor(); + + assertThat(cqListenerContainer.getContinuousQueries()).hasSize(queries.size()); + + cqListenerContainer.initExecutor(); + cqListenerContainer.destroy(); + + assertThat(cqListenerContainer.getContinuousQueries()).isEmpty(); + + verify(mockQueryOne, times(1)).isClosed(); + verify(mockQueryOne, times(1)).close(); + verify(mockQueryTwo, times(1)).isClosed(); + verify(mockQueryTwo, never()).close(); + verify(mockQueryThree, times(1)).isClosed(); + verify(mockQueryThree, times(1)).close(); + verify(mockQueryFour, times(1)).isClosed(); + verify(mockQueryFour, times(1)).close(); + verify(mockDisposableExecutorBean, times(1)).destroy(); + } + + interface DisposableExecutorBean extends Executor, DisposableBean { } + }