SGF-668 - Add Annotation configuration support for Continuous Queries.

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<ContinuousQueryListenerContainerConfigurer> 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.<Integer>getNumber("phase"));
+			setPoolName(enableContinuousQueriesAttributes.getString("poolName"));
+			setTaskExecutorBeanName(enableContinuousQueriesAttributes.getString("taskExecutorBeanName"));
+		}
+	}
+
+	@Bean
+	public BeanPostProcessor continuousQueryBeanPostProcessor() {
+
+		return new BeanPostProcessor() {
+
+			private ContinuousQueryListenerContainer container;
+
+			private List<ContinuousQueryDefinition> 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<ContinuousQueryDefinition> 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<ContinuousQueryListenerContainerConfigurer> resolveContinuousQueryListenerContainerConfigurers() {
+
+		return Optional.ofNullable(this.configurers)
+			.filter(configurers -> !configurers.isEmpty())
+			.orElseGet(() ->
+				Optional.of(this.beanFactory())
+					.filter(beanFactory -> beanFactory instanceof ListableBeanFactory)
+					.map(beanFactory -> {
+
+						Map<String, ContinuousQueryListenerContainerConfigurer> beansOfType =
+							((ListableBeanFactory) beanFactory).getBeansOfType(ContinuousQueryListenerContainerConfigurer.class,
+								true, true);
+
+						return nullSafeMap(beansOfType).values().stream().collect(Collectors.toList());
+					})
+					.orElseGet(Collections::emptyList)
+			);
+	}
+
+	protected Optional<ErrorHandler> resolveErrorHandler() {
+
+		return Optional.ofNullable(getErrorHandlerBeanName())
+			.filter(StringUtils::hasText)
+			.map(errorHandlerBeanName -> beanFactory().getBean(errorHandlerBeanName, ErrorHandler.class));
+	}
+
+	protected Optional<Integer> resolvePhase() {
+		return Optional.of(getPhase()).filter(phase -> phase != 0);
+	}
+
+	protected Optional<String> resolvePoolName() {
+		return Optional.ofNullable(getPoolName()).filter(StringUtils::hasText);
+	}
+
+	protected Optional<Executor> 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;
+	}
+}

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);
+
+}

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 "";
+
+}

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);
 	}

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<ContinuousQueryListenerContainerConfigurer> 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<CqQuery> 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<ContinuousQueryListenerContainerConfigurer> 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<ContinuousQueryListenerContainerConfigurer> configurers) {
+		this.cqListenerContainerConfigurers = Optional.ofNullable(configurers).orElseGet(Collections::emptyList);
+	}
+
+	/**
+	 * Returns a <a href="https://en.wikipedia.org/wiki/Composite_pattern">Composite</a> 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.

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.
  *
- * <p>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.
  *
- * <p>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}.</p>
+ * <p>Modeled as much as possible after the JMS MessageListenerAdapter in the core Spring Framework.
  *
- * <p>Find below some examples of method signatures compliant with this
- * adapter class. This first example handles all <code>CqEvent</code> types
- * and gets passed the contents of each <code>event</code> type as an
- * argument.</p>
+ * <p>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}.</p>
+ *
+ * <p>Find below some examples of method signatures compliant with this adapter class.
+ *
+ * This first example handles all <code>CqEvent</code> types and gets passed the contents of each
+ * <code>event</code> type as an argument.</p>
  *
  * <pre class="code">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);
  * }</pre>
@@ -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.
-	 * <p>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.
-	 * <p>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.
+	 *
+	 * <p>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<Method> methods;
+		private final List<Method> 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<Method>();
+			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;
 		}
 	}
-
 }

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();
+
+}