Fixes JIRA issues SGF-271 - <cq-listener-container> element is missing the 'auto-startup' attribute, SGF-275 - <cq-listener-container> element's phase attribute is ignored, SGF-278 - ContinuousQueryListenerContainer class's 'taskExecutor' property is not set properly by the GemfireListenerContainerParser, SGF-279 - <cq-listener-container> element is missing the 'error-handler' attribute, and finally, SGF-280 - the ContinuousQueryListenerContainer class is not Thread-safe.

src/main/java/org/springframework/data/gemfire/config/GemfireListenerContainerParser.java

@@ -34,8 +34,10 @@ import org.w3c.dom.Element;
  * Parser for SGF <code>&lt;cq-listener-container&gt;</code> element.
  * 
  * @author Costin Leau
+ * @author John Blum
  */
 class GemfireListenerContainerParser extends AbstractSingleBeanDefinitionParser {
+
 	@Override
 	protected Class<ContinuousQueryListenerContainer> getBeanClass(Element element) {
 		return ContinuousQueryListenerContainer.class;
@@ -43,19 +45,21 @@ class GemfireListenerContainerParser extends AbstractSingleBeanDefinitionParser
 
 	@Override
 	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
-
+		ParsingUtils.setPropertyReference(element, builder, "cache", "cache");
+		ParsingUtils.setPropertyValue(element, builder, "auto-startup");
 		ParsingUtils.setPropertyValue(element, builder, "phase");
 		ParsingUtils.setPropertyValue(element, builder, "pool-name");
-		ParsingUtils.setPropertyReference(element, builder, "cache","cache");
-		ParsingUtils.setPropertyReference(element, builder, "task-executor","task-executor");
+		ParsingUtils.setPropertyReference(element, builder, "error-handler", "errorHandler");
+		ParsingUtils.setPropertyReference(element, builder, "task-executor", "taskExecutor");
 
-		// parse nested listeners
-		List<Element> listDefs = DomUtils.getChildElementsByTagName(element, "listener");
+		// parse nested Continuous Query Listeners
+		List<Element> listenerElements = DomUtils.getChildElementsByTagName(element, "listener");
 
-		if (!listDefs.isEmpty()) {
-			ManagedSet<BeanDefinition> listeners = new ManagedSet<BeanDefinition>(listDefs.size());
-			for (Element listElement : listDefs) {
-				listeners.add(parseListener(listElement));
+		if (!listenerElements.isEmpty()) {
+			ManagedSet<BeanDefinition> listeners = new ManagedSet<BeanDefinition>(listenerElements.size());
+
+			for (Element listenerElement : listenerElements) {
+				listeners.add(parseListener(listenerElement));
 			}
 
 			builder.addPropertyValue("queryListeners", listeners);
@@ -69,30 +73,36 @@ class GemfireListenerContainerParser extends AbstractSingleBeanDefinitionParser
 	 * @return
 	 */
 	private BeanDefinition parseListener(Element element) {
+		BeanDefinitionBuilder continuousQueryListenerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
+			ContinuousQueryListenerAdapter.class);
 
-		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ContinuousQueryListenerAdapter.class);
-		builder.addConstructorArgReference(element.getAttribute("ref"));
+		continuousQueryListenerBuilder.addConstructorArgReference(element.getAttribute("ref"));
 
-		String attr = element.getAttribute("method");
-		if (StringUtils.hasText(attr)) {
-			builder.addPropertyValue("defaultListenerMethod", attr);
+		String attribute = element.getAttribute("method");
+
+		if (StringUtils.hasText(attribute)) {
+			continuousQueryListenerBuilder.addPropertyValue("defaultListenerMethod", attribute);
 		}
 
-		BeanDefinitionBuilder defBuilder = BeanDefinitionBuilder.genericBeanDefinition(ContinuousQueryDefinition.class);
+		BeanDefinitionBuilder continuousQueryBuilder = BeanDefinitionBuilder.genericBeanDefinition(
+			ContinuousQueryDefinition.class);
 
-		attr = element.getAttribute("name");
-		if (StringUtils.hasText(attr)) {
-			defBuilder.addConstructorArgValue(attr);
+		attribute = element.getAttribute("name");
+
+		if (StringUtils.hasText(attribute)) {
+			continuousQueryBuilder.addConstructorArgValue(attribute);
 		}
 
-		defBuilder.addConstructorArgValue(element.getAttribute("query"));
-		defBuilder.addConstructorArgValue(builder.getBeanDefinition());
+		continuousQueryBuilder.addConstructorArgValue(element.getAttribute("query"));
+		continuousQueryBuilder.addConstructorArgValue(continuousQueryListenerBuilder.getBeanDefinition());
 
-		attr = element.getAttribute("durable");
-		if (StringUtils.hasText(attr)) {
-			defBuilder.addConstructorArgValue(attr);
+		attribute = element.getAttribute("durable");
+
+		if (StringUtils.hasText(attribute)) {
+			continuousQueryBuilder.addConstructorArgValue(attribute);
 		}
 
-		return defBuilder.getBeanDefinition();
+		return continuousQueryBuilder.getBeanDefinition();
 	}
-}
+
+}

src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryDefinition.java

@@ -22,17 +22,23 @@ import org.springframework.util.Assert;
 import com.gemstone.gemfire.cache.query.CqQuery;
 
 /**
- * Basic holder class for defining an {@link CqQuery}. Useful for configuring GemFire {@link CqQuery}s through XML
- * and or JavaBeans means.
+ * Basic holder class for defining an {@link CqQuery}. Useful for configuring GemFire {@link CqQuery}s by mean of
+ * XML or using JavaBeans.
  * 
  * @author Costin Leau
+ * @author John Blum
+ * @see org.springframework.beans.factory.InitializingBean
  */
+@SuppressWarnings("unused")
 public class ContinuousQueryDefinition implements InitializingBean {
 
-	private String name = null, query = null;
-	private ContinuousQueryListener listener = null;
 	private boolean durable = false;
 
+	private ContinuousQueryListener listener;
+
+	private String name;
+	private String query;
+
 	public ContinuousQueryDefinition() {
 	}
 
@@ -60,35 +66,44 @@ public class ContinuousQueryDefinition implements InitializingBean {
 	}
 
 	public void afterPropertiesSet() {
-		Assert.hasText(query, "a non-empty query is required");
-		Assert.notNull(listener, "a non- null listener is required");
+		Assert.hasText(query, "A non-empty query is required.");
+		Assert.notNull(listener, "A non-null listener is required.");
 	}
 
 	/**
-	 * @return the name
+	 * Determines whether the CQ is durable.
+	 *
+	 * @return a boolean indicating if the CQ is durable.
+	 */
+	public boolean isDurable() {
+		return durable;
+	}
+
+	/**
+	 * Gets the name for the CQ.
+	 *
+	 * @return a String name for the CQ.
 	 */
 	public String getName() {
 		return name;
 	}
 
 	/**
-	 * @return the query
+	 * Gets the query string that will be executed for the CQ.
+	 *
+	 * @return a String value with the query to be executed for the CQ.
 	 */
 	public String getQuery() {
 		return query;
 	}
 
 	/**
-	 * @return the listener
+	 * The CQ Listener receiving events and notifications with changes from the CQ.
+	 *
+	 * @return the listener to be registered for the CQ.
 	 */
 	public ContinuousQueryListener getListener() {
 		return listener;
 	}
 
-	/**
-	 * @return the durable
-	 */
-	public boolean isDurable() {
-		return durable;
-	}
-}
+}

src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListener.java

@@ -19,11 +19,19 @@ package org.springframework.data.gemfire.listener;
 import com.gemstone.gemfire.cache.query.CqEvent;
 
 /**
- * Continuous query listener. 
+ * Continuous Query (CQ) listener listening for events and notifications by a GemFire Continuous Query (CQ).
  * 
  * @author Costin Leau
+ * @author John Blum
  */
 public interface ContinuousQueryListener {
 
+	/**
+	 * Action performed by the listener when notified of a CQ event.
+	 *
+	 * @param event the event from the CQ.
+	 * @see com.gemstone.gemfire.cache.query.CqEvent
+	 */
 	void onEvent(CqEvent event);
+
 }

src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java

@@ -17,7 +17,9 @@
 package org.springframework.data.gemfire.listener;
 
 import java.util.LinkedHashSet;
+import java.util.Queue;
 import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
 
 import org.apache.commons.logging.Log;
@@ -44,19 +46,438 @@ import com.gemstone.gemfire.cache.query.CqListener;
 import com.gemstone.gemfire.cache.query.CqQuery;
 import com.gemstone.gemfire.cache.query.QueryException;
 import com.gemstone.gemfire.cache.query.QueryService;
-import com.gemstone.gemfire.internal.concurrent.ConcurrentHashSet;
 
 /**
  * Container providing asynchronous behaviour for GemFire continuous queries.
- * 
+ *
  * @author Costin Leau
+ * @author John Blum
+ * @see org.springframework.beans.factory.BeanNameAware
+ * @see org.springframework.beans.factory.DisposableBean
+ * @see org.springframework.beans.factory.InitializingBean
+ * @see org.springframework.context.SmartLifecycle
+ * @see org.springframework.core.task.SimpleAsyncTaskExecutor
+ * @see org.springframework.core.task.TaskExecutor
+ * @see com.gemstone.gemfire.cache.RegionService
+ * @see com.gemstone.gemfire.cache.client.Pool
+ * @see com.gemstone.gemfire.cache.client.PoolManager
+ * @see com.gemstone.gemfire.cache.query.CqEvent
+ * @see com.gemstone.gemfire.cache.query.CqListener
+ * @see com.gemstone.gemfire.cache.query.CqQuery
+ * @see com.gemstone.gemfire.cache.query.QueryService
  */
-public class ContinuousQueryListenerContainer implements InitializingBean, DisposableBean, BeanNameAware, SmartLifecycle {
+@SuppressWarnings("unused")
+public class ContinuousQueryListenerContainer implements BeanNameAware, InitializingBean, DisposableBean, SmartLifecycle {
+
+	// Default Thread name prefix is "ContinuousQueryListenerContainer-".
+	public static final String DEFAULT_THREAD_NAME_PREFIX = String.format("%1$s-", ClassUtils.getShortName(
+		ContinuousQueryListenerContainer.class));
+
+	private boolean autoStartup = true;
+
+	private volatile boolean initialized = false;
+	private volatile boolean manageExecutor = false;
+	private volatile boolean running = false;
+
+	private int phase = Integer.MAX_VALUE;
+
+	private ErrorHandler errorHandler;
+
+	private Executor taskExecutor;
+
+	protected final Log logger = LogFactory.getLog(getClass());
+
+	private Queue<CqQuery> continuousQueries = new ConcurrentLinkedQueue<CqQuery>();
+
+	private QueryService queryService;
+
+	private Set<ContinuousQueryDefinition> continuousQueryDefinitions = new LinkedHashSet<ContinuousQueryDefinition>();
+
+	private String beanName;
+	private String poolName;
+
+	public void afterPropertiesSet() {
+		initQueryService();
+		initExecutor();
+		initContinuousQueries(continuousQueryDefinitions);
+		initialized = true;
+
+		if (isAutoStartup()) {
+			start();
+		}
+	}
+
+	private void initQueryService() {
+		if (StringUtils.hasText(poolName)) {
+			Pool pool = PoolManager.find(poolName);
+			Assert.notNull(pool, String.format("No GemFire Pool with name '%1$s' was found.", poolName));
+			queryService = pool.getQueryService();
+		}
+	}
+
+	private void initExecutor() {
+		if (taskExecutor == null) {
+			manageExecutor = true;
+			taskExecutor = createDefaultTaskExecutor();
+		}
+	}
+
+	/**
+	 * Creates a default TaskExecutor. Called if no explicit TaskExecutor has been configured.
+	 * <p>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.</p>
+	 *
+	 * @see org.springframework.core.task.SimpleAsyncTaskExecutor#SimpleAsyncTaskExecutor(String)
+	 */
+	protected TaskExecutor createDefaultTaskExecutor() {
+		return new SimpleAsyncTaskExecutor(beanName != null ? String.format("%1$s-", beanName)
+			: DEFAULT_THREAD_NAME_PREFIX);
+	}
+
+	private void initContinuousQueries(Set<ContinuousQueryDefinition> continuousQueryDefinitions) {
+		// stop the continuous query listener container if currently running...
+		if (isRunning()) {
+			stop();
+		}
+
+		// close any existing continuous queries...
+		closeQueries();
+
+		// 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 ex) {
+			throw new GemfireQueryException(String.format("Could not execute query '%1$s'; state is '%2$s'.",
+				cq.getName(), cq.getState()), ex);
+		}
+		catch (RuntimeException ex) {
+			throw new GemfireQueryException(String.format("Could not execute query '%1$s'; state is '%2$s'.",
+				cq.getName(), cq.getState()), ex);
+		}
+	}
+
+	public synchronized void stop() {
+		if (isRunning()) {
+			doStop();
+			running = false;
+		}
+
+		if (logger.isDebugEnabled()) {
+			logger.debug("Stopped ContinuousQueryListenerContainer");
+		}
+	}
+
+	public void stop(final Runnable callback) {
+		stop();
+		callback.run();
+	}
+
+	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);
+			}
+		}
+	}
+
+	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)
+	 * but not shutdown yet.
+	 *
+	 * @return a boolean indicating whether the container is active.
+	 */
+	public final boolean isActive() {
+		return initialized;
+	}
+
+	/**
+	 * Determines whether the container has be started and is currently running.
+	 *
+	 * @return a boolean value indicating whether the container has been started and is currently running.
+	 */
+	public synchronized boolean isRunning() {
+		return running;
+	}
+
+	/**
+	 * Determines whether this CQ listener container will automatically start on startup.
+	 *
+	 * @return a boolean value indicating whether this CQ listener container automatically starts.
+	 * @see org.springframework.context.SmartLifecycle#isAutoStartup()
+	 */
+	public boolean isAutoStartup() {
+		return autoStartup;
+	}
+
+	/**
+	 * Sets whether the CQ listener container should automatically start on startup.
+	 *
+	 * @param autoStartup a boolean value indicating whether this CQ listener container should automatically start.
+	 */
+	public void setAutoStartup(final boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	/**
+	 * Set the name of the bean in the bean factory that created this bean.
+	 * <p>Invoked after population of normal bean properties but before an
+	 * init callback such as {@link InitializingBean#afterPropertiesSet()}
+	 * or a custom init-method.</p>
+	 *
+	 * @param name the name of the bean in the factory.
+	 */
+	public void setBeanName(String name) {
+		this.beanName = name;
+	}
+
+	/**
+	 * Set the underlying RegionService (GemFire Cache) used for registering Queries.
+	 *
+	 * @param cache the RegionService (GemFire Cache) used for registering Queries.
+	 * @see com.gemstone.gemfire.cache.RegionService
+	 */
+	public void setCache(RegionService cache) {
+		setQueryService(cache.getQueryService());
+	}
+
+	/**
+	 * Set an ErrorHandler to be invoked in case of any uncaught exceptions thrown
+	 * while processing a event. By default there will be <b>no</b> ErrorHandler
+	 * so that error-level logging is the only result.
+	 */
+	public void setErrorHandler(ErrorHandler errorHandler) {
+		this.errorHandler = errorHandler;
+	}
+
+	/**
+	 * Gets the phase in which this CQ listener container will start in the Spring container.
+	 *
+	 * @return the phase value of this CQ listener container.
+	 * @see org.springframework.context.Phased#getPhase()
+	 */
+	public int getPhase() {
+		return phase;
+	}
+
+	/**
+	 * Sets the phase in which this CQ listener container will start in the Spring container.
+	 *
+	 * @param phase the phase value of this CQ listener container.
+	 */
+	public void setPhase(final int phase) {
+		this.phase = phase;
+	}
+
+	/**
+	 * Set the name of the {@link Pool} used for performing the queries by this container.
+	 *
+	 * @param poolName the name of the pool to be used by the container
+	 */
+	public void setPoolName(String poolName) {
+		this.poolName = poolName;
+	}
+
+	/**
+	 * Attaches the given query definitions.
+	 *
+	 * @param queries set of queries
+	 */
+	public void setQueryListeners(Set<ContinuousQueryDefinition> queries) {
+		continuousQueryDefinitions.clear();
+		continuousQueryDefinitions.addAll(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).
+	 * @see com.gemstone.gemfire.cache.query.QueryService
+	 */
+	public void setQueryService(QueryService service) {
+		this.queryService = service;
+	}
+
+	/**
+	 * Sets the Task Executor used for running the event listeners when messages are received.
+	 * If no task executor is set, an instance of {@link SimpleAsyncTaskExecutor} will be used by default.
+	 * The task executor can be adjusted depending on the work done by the listeners and the number of
+	 * messages coming in.
+	 *
+	 * @param taskExecutor The Task Executor used to run event listeners when query results messages are received.
+	 * @see java.util.concurrent.Executor
+	 */
+	public void setTaskExecutor(Executor taskExecutor) {
+		this.taskExecutor = taskExecutor;
+	}
+
+	/**
+	 * 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.
+	 *
+	 * @param definition Continuous Query (CQ) definition
+	 * @see org.springframework.data.gemfire.listener.ContinuousQueryDefinition
+	 * @see #doAddListener(ContinuousQueryDefinition)
+	 */
+	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 (RuntimeException ex) {
+			throw new GemfireQueryException("Cannot create query ", ex);
+		}
+		catch (QueryException ex) {
+			throw new GemfireQueryException("Cannot create query ", ex);
+		}
+	}
+
+	private void dispatchEvent(final ContinuousQueryListener listener, final CqEvent event) {
+		taskExecutor.execute(new Runnable() {
+			public void run() {
+				executeListener(listener, event);
+			}
+		});
+	}
+
+	/**
+	 * Execute the specified listener.
+	 *
+	 * @see #handleListenerException
+	 */
+	protected void executeListener(ContinuousQueryListener listener, CqEvent event) {
+		try {
+			listener.onEvent(event);
+		}
+		catch (Throwable ex) {
+			handleListenerException(ex);
+		}
+	}
+
+	/**
+	 * Handle the given exception that arose during listener execution.
+	 * <p>The default implementation logs the exception at error level.
+	 * This can be overridden in subclasses.
+	 *
+	 * @param e the exception to handle
+	 */
+	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);
+		}
+	}
+
+	/**
+	 * 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);
+		}
+		else if (logger.isWarnEnabled()) {
+			logger.warn("Execution of the CQ event listener failed, and no ErrorHandler has been set.", e);
+		}
+	}
 
 	private class EventDispatcherAdapter implements CqListener {
+
 		private final ContinuousQueryListener delegate;
 
-		EventDispatcherAdapter(ContinuousQueryListener delegate) {
+		private EventDispatcherAdapter(final ContinuousQueryListener delegate) {
 			this.delegate = delegate;
 		}
 
@@ -72,349 +493,4 @@ public class ContinuousQueryListenerContainer implements InitializingBean, Dispo
 		}
 	}
 
-	/** Logger available to subclasses */
-	protected final Log logger = LogFactory.getLog(getClass());
-
-	/**
-	 * Default thread name prefix: "ContinousQueryListenerContainer-".
-	 */
-	public static final String DEFAULT_THREAD_NAME_PREFIX = ClassUtils.getShortName(ContinuousQueryListenerContainer.class)
-			+ "-";
-
-	private Executor subscriptionExecutor;
-	private Executor taskExecutor;
-	private String beanName;
-	private ErrorHandler errorHandler;
-
-	// whether the container is running (or not)
-	private volatile boolean running = false;
-	// whether the container has been initialized
-	private volatile boolean initialized = false;
-	private volatile boolean manageExecutor = false;
-
-	private Set<ContinuousQueryDefinition> defs = new LinkedHashSet<ContinuousQueryDefinition>();
-	private Set<CqQuery> queries = new ConcurrentHashSet<CqQuery>();
-
-	private QueryService queryService;
-	private String poolName;
-
-
-	public void afterPropertiesSet() {
-		if (taskExecutor == null) {
-			manageExecutor = true;
-			taskExecutor = createDefaultTaskExecutor();
-		}
-
-		if (subscriptionExecutor == null) {
-			subscriptionExecutor = taskExecutor;
-		}
-
-		if (StringUtils.hasText(poolName)) {
-			Pool pool = PoolManager.find(poolName);
-			Assert.notNull(pool, "No pool named [" + poolName + "] found");
-			queryService = pool.getQueryService();
-		}
-
-		initMapping(defs);
-		initialized = true;
-
-		start();
-	}
-
-	/**
-	 * Creates a default TaskExecutor. Called if no explicit TaskExecutor has been specified.
-	 * <p>The default implementation builds a {@link org.springframework.core.task.SimpleAsyncTaskExecutor}
-	 * with the specified bean name (or the class name, if no bean name specified) as thread name prefix.
-	 * @see org.springframework.core.task.SimpleAsyncTaskExecutor#SimpleAsyncTaskExecutor(String)
-	 */
-	protected TaskExecutor createDefaultTaskExecutor() {
-		String threadNamePrefix = (beanName != null ? beanName + "-" : DEFAULT_THREAD_NAME_PREFIX);
-		return new SimpleAsyncTaskExecutor(threadNamePrefix);
-	}
-
-	public void destroy() throws Exception {
-		initialized = false;
-
-		stop();
-		closeQueries();
-
-		if (manageExecutor) {
-			if (taskExecutor instanceof DisposableBean) {
-				((DisposableBean) taskExecutor).destroy();
-
-				if (logger.isDebugEnabled()) {
-					logger.debug("Stopped internally-managed task executor");
-				}
-			}
-		}
-	}
-
-	public boolean isAutoStartup() {
-		return true;
-	}
-
-	public void stop(Runnable callback) {
-		stop();
-		callback.run();
-	}
-
-	public int getPhase() {
-		// start the latest
-		return Integer.MAX_VALUE;
-	}
-
-	public boolean isRunning() {
-		return running;
-	}
-
-	public void start() {
-		if (!running) {
-			running = true;
-
-			doStart();
-
-			if (logger.isDebugEnabled()) {
-				logger.debug("Started ContinousQueryListenerContainer");
-			}
-		}
-	}
-
-	public void stop() {
-		if (running) {
-			running = false;
-			doStop();
-		}
-
-		if (logger.isDebugEnabled()) {
-			logger.debug("Stopped ContinousQueryListenerContainer");
-		}
-	}
-
-	private void doStart() {
-		for (CqQuery cq : queries) {
-			executeQuery(cq);
-		}
-	}
-
-	private void doStop() {
-		for (CqQuery cq : queries) {
-			try {
-				cq.stop();
-			} catch (RuntimeException ex) {
-				logger.warn("Cannot stop query", ex);
-			} catch (QueryException ex) {
-				logger.warn("Cannot stop query", ex);
-			}
-		}
-	}
-
-	private void closeQueries() {
-		for (CqQuery cq : queries) {
-			try {
-				if (!cq.isClosed()) {
-					cq.close();
-				}
-			} catch (QueryException ex) {
-				logger.warn("Cannot close query", ex);
-			} catch (RuntimeException ex) {
-				logger.warn("Cannot close query", ex);
-			}
-		}
-
-		queries.clear();
-	}
-
-	/**
-	 * Execute the specified listener.
-	 * 
-	 * @see #handleListenerException
-	 */
-	protected void executeListener(ContinuousQueryListener listener, CqEvent event) {
-		try {
-			listener.onEvent(event);
-		} catch (Throwable ex) {
-			handleListenerException(ex);
-		}
-	}
-
-	/**
-	 * Return whether this container is currently active,
-	 * that is, whether it has been set up but not shut down yet.
-	 */
-	public final boolean isActive() {
-		return initialized;
-	}
-
-	/**
-	 * Handle the given exception that arose during listener execution.
-	 * <p>The default implementation logs the exception at error level.
-	 * This can be overridden in subclasses.
-	 * @param ex the exception to handle
-	 */
-	protected void handleListenerException(Throwable ex) {
-		if (isActive()) {
-			// Regular case: failed while active.
-			// Invoke ErrorHandler if available.
-			invokeErrorHandler(ex);
-		}
-		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", ex);
-		}
-	}
-
-	/**
-	 * Invoke the registered ErrorHandler, if any. Log at error level otherwise.
-	 * @param ex the uncaught error that arose during event processing.
-	 * @see #setErrorHandler
-	 */
-	protected void invokeErrorHandler(Throwable ex) {
-		if (this.errorHandler != null) {
-			this.errorHandler.handleError(ex);
-		}
-		else if (logger.isWarnEnabled()) {
-			logger.warn("Execution of the CQ event listener failed, and no ErrorHandler has been set.", ex);
-		}
-	}
-	
-	public void setBeanName(String name) {
-		this.beanName = name;
-	}
-
-	/**
-	 * Sets the task executor used for running the event listeners when messages are received.
-	 * If no task executor is set, an instance of {@link SimpleAsyncTaskExecutor} will be used by default.
-	 * The task executor can be adjusted depending on the work done by the listeners and the number of 
-	 * messages coming in.
-	 * 
-	 * @param taskExecutor The taskExecutor to set.
-	 */
-	public void setTaskExecutor(Executor taskExecutor) {
-		this.taskExecutor = taskExecutor;
-	}
-
-	/**
-	 * Set an ErrorHandler to be invoked in case of any uncaught exceptions thrown
-	 * while processing a event. By default there will be <b>no</b> ErrorHandler
-	 * so that error-level logging is the only result.
-	 */
-	public void setErrorHandler(ErrorHandler errorHandler) {
-		this.errorHandler = errorHandler;
-	}
-
-	/**
-	 * Set the underlying cache used for registering queries.
-	 * 
-	 * @param cache cache used for registering queries
-	 */
-	public void setCache(RegionService cache) {
-		this.queryService = cache.getQueryService();
-	}
-
-	/**
-	 * Set the query service to be used by this container.
-	 * 
-	 * @param service query service used by the container
-	 */
-	public void setQueryService(QueryService service) {
-		this.queryService = service;
-	}
-
-	/**
-	 * Set the name of the {@link Pool} used for performing the queries by this container.
-	 * 
-	 * @param poolName the name of the pool to be used by the container
-	 */
-	public void setPoolName(String poolName) {
-		this.poolName = poolName;
-	}
-
-	/**
-	 * Attaches the given query definitions.
-	 * 
-	 * @param queries set of queries
-	 */
-	public void setQueryListeners(Set<ContinuousQueryDefinition> queries) {
-		defs.clear();
-		defs.addAll(queries);
-	}
-
-	/**
-	 * Adds a query definition to the (potentially running) container. If the container is running,
-	 * the listener starts receiving (matching) messages as soon as possible.
-	 * 
-	 * @param cqQuery cqQuery definition
-	 */
-	public void addListener(ContinuousQueryDefinition cqQuery) {
-		doAddListener(cqQuery);
-	}
-
-	private void initMapping(Set<ContinuousQueryDefinition> queryDefinitions) {
-		// stop the listener if currently running
-		if (isRunning()) {
-			stop();
-		}
-
-		closeQueries();
-
-		for (ContinuousQueryDefinition def : queryDefinitions) {
-			addCQuery(def);
-		}
-
-		// resume activity
-		if (initialized) {
-			start();
-		}
-	}
-
-	private void doAddListener(ContinuousQueryDefinition def) {
-		CqQuery cq = addCQuery(def);
-
-		if (isRunning()) {
-			executeQuery(cq);
-		}
-	}
-
-	private CqQuery addCQuery(ContinuousQueryDefinition def) {
-		try {
-			CqAttributesFactory caf = new CqAttributesFactory();
-			caf.addCqListener(new EventDispatcherAdapter(def.getListener()));
-			CqAttributes attr = caf.create();
-
-			CqQuery cq = null;
-
-			if (StringUtils.hasText(def.getName())) {
-				cq = queryService.newCq(def.getName(), def.getQuery(), attr, def.isDurable());
-			}
-			else {
-				cq = queryService.newCq(def.getQuery(), attr, def.isDurable());
-			}
-
-			queries.add(cq);
-			return cq;
-		} catch (RuntimeException ex) {
-			throw new GemfireQueryException("Cannot create query ", ex);
-		} catch (QueryException ex) {
-			throw new GemfireQueryException("Cannot create query ", ex);
-		}
-	}
-
-	private void executeQuery(CqQuery cq) {
-		try {
-			cq.execute();
-		} catch (QueryException ex) {
-			throw new GemfireQueryException("Cannot execute query", ex);
-		} catch (RuntimeException ex) {
-			throw new GemfireQueryException("Cannot execute query", ex);
-		}
-	}
-
-	private void dispatchEvent(final ContinuousQueryListener listener, final CqEvent event) {
-		taskExecutor.execute(new Runnable() {
-			public void run() {
-				executeListener(listener, event);
-			}
-		});
-	}
-}
+}

src/main/java/org/springframework/data/gemfire/listener/adapter/ContinuousQueryListenerAdapter.java

@@ -43,7 +43,7 @@ import com.gemstone.gemfire.cache.query.CqQuery;
  * Allows listener methods to operate on event content types, completely
  * independent from the GemFire API.
  * 
- * Modeled as much as possible after the JMS MessageListenerAdapter in
+ * <p>Modeled as much as possible after the JMS MessageListenerAdapter in
  * Spring Framework.
  *
  * <p>By default, the content of incoming GemFire events gets extracted before
@@ -69,129 +69,21 @@ import com.gemstone.gemfire.cache.query.CqQuery;
  * @author Juergen Hoeller
  * @author Costin Leau
  * @author Oliver Gierke
+ * @author John Blum
  */
 public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
 
-	private class MethodInvoker {
-		private final Object delegate;
-		List<Method> methods;
-
-		MethodInvoker(Object delegate, final String methodName) {
-			this.delegate = delegate;
-
-			Class<?> c = delegate.getClass();
-
-			methods = new ArrayList<Method>();
-
-			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) {
-					if (Modifier.isPublic(method.getModifiers()) && methodName.equals(method.getName())) {
-
-						// check out the arguments
-						Class<?>[] parameterTypes = method.getParameterTypes();
-						int objects = 0;
-						int operations = 0;
-
-						if (parameterTypes.length > 0) {
-							for (Class<?> paramType : parameterTypes) {
-
-								if (Object.class.equals(paramType)) {
-									objects++;
-									if (objects > 2) {
-										return false;
-									}
-								}
-								else if (Operation.class.equals(paramType)) {
-									operations++;
-									if (operations > 2) {
-										return false;
-									}
-								}
-								else if (CqEvent.class.equals(paramType)) {
-								}
-								else if (Throwable.class.equals(paramType)) {
-								}
-								else if (byte[].class.equals(paramType)) {
-								}
-								else if (CqQuery.class.equals(paramType)) {
-								}
-								else {
-									return false;
-								}
-							}
-							return true;
-						}
-					}
-					return false;
-				}
-			});
-
-			Assert.isTrue(!methods.isEmpty(), "Cannot find a suitable method named [" + c.getName() + "#" + methodName
-					+ "] - is the method public and has the proper arguments?");
-		}
-
-		void invoke(CqEvent event) throws InvocationTargetException, IllegalAccessException {
-
-			for (Method m : methods) {
-				Class<?>[] types = m.getParameterTypes();
-				Object[] args = new Object[types.length];
-
-				boolean value = false;
-				boolean query = false;
-
-				for (int i = 0; i < types.length; i++) {
-					Class<?> paramType = types[i];
-
-					if (Object.class.equals(paramType)) {
-						args[i] = (!value ? event.getKey() : event.getNewValue());
-						value = true;
-					}
-					else if (Operation.class.equals(paramType)) {
-						args[i] = (!query ? event.getBaseOperation() : event.getQueryOperation());
-						query = true;
-					}
-					else if (CqEvent.class.equals(paramType)) {
-						args[i] = event;
-					}
-					else if (Throwable.class.equals(paramType)) {
-						args[i] = event.getThrowable();
-					}
-					else if (byte[].class.equals(paramType)) {
-						args[i] = event.getDeltaValue();
-					}
-					else if (CqQuery.class.equals(paramType)) {
-						args[i] = event.getCq();
-					}
-				}
-
-				m.invoke(delegate, args);
-			}
-		}
-	}
-
-
-	/**
-	 * Out-of-the-box value for the default listener method: "handleEvent".
-	 */
+	// Out-of-the-box value for the default listener handler method "handleEvent".
 	public static final String ORIGINAL_DEFAULT_LISTENER_METHOD = "handleEvent";
 
-
-	/** Logger available to subclasses */
 	protected final Log logger = LogFactory.getLog(getClass());
 
+	private MethodInvoker invoker;
+
 	private Object delegate;
 
 	private String defaultListenerMethod = ORIGINAL_DEFAULT_LISTENER_METHOD;
 
-	private MethodInvoker invoker;
-
 	/**
 	 * Create a new {@link ContinuousQueryListenerAdapter} with default settings.
 	 */
@@ -208,7 +100,6 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
 		setDelegate(delegate);
 	}
 
-
 	/**
 	 * Set a target object to delegate events listening to.
 	 * Specified listener methods have to be present on this target object.
@@ -219,7 +110,7 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
 	 * @param delegate delegate object
 	 */
 	public void setDelegate(Object delegate) {
-		Assert.notNull(delegate, "Delegate must not be null");
+		Assert.notNull(delegate, "The delegate must not be null.");
 		this.delegate = delegate;
 		this.invoker = null;
 	}
@@ -262,8 +153,7 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
 	 */
 	public void onEvent(CqEvent event) {
 		try {
-
-			// Check whether the delegate is a ContinuousQueryListener impl itself.
+			// Check whether the delegate is a ContinuousQueryListener implementation itself.
 			// In that case, the adapter will simply act as a pass-through.
 			if (delegate != this) {
 				if (delegate instanceof ContinuousQueryListener) {
@@ -272,16 +162,18 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
 				}
 			}
 
-			// Regular case: find a handler method reflectively.
+			// Other case... find the listener handler method reflectively.
 			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.");
+			}
+
 			if (invoker == null) {
 				invoker = new MethodInvoker(delegate, methodName);
 			}
-			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.");
-			}
 
 			invokeListenerMethod(event, methodName);
 		} catch (Throwable th) {
@@ -289,15 +181,6 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
 		}
 	}
 
-	/**
-	 * Handle the given exception that arose during listener execution.
-	 * The default implementation logs the exception at error level.
-	 * @param ex the exception to handle
-	 */
-	protected void handleListenerException(Throwable ex) {
-		logger.error("Listener execution failed", ex);
-	}
-
 	/**
 	 * Determine the name of the listener method that is supposed to
 	 * handle the given event.
@@ -307,10 +190,20 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
 	 * @return the name of the listener method (never <code>null</code>)
 	 * @see #setDefaultListenerMethod
 	 */
+	@SuppressWarnings("unused")
 	protected String getListenerMethodName(CqEvent event) {
 		return getDefaultListenerMethod();
 	}
 
+	/**
+	 * Handle the given exception that arose during listener execution.
+	 * The default implementation logs the exception at error level.
+	 * @param ex the exception to handle
+	 */
+	protected void handleListenerException(Throwable ex) {
+		logger.error("Listener execution failed...", ex);
+	}
+
 	/**
 	 * Invoke the specified listener method.
 	 * @param event the event arguments to be passed in
@@ -320,17 +213,124 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
 	protected void invokeListenerMethod(CqEvent event, String methodName) {
 		try {
 			invoker.invoke(event);
-		} catch (InvocationTargetException ex) {
-			Throwable targetEx = ex.getTargetException();
-			if (targetEx instanceof DataAccessException) {
-				throw (DataAccessException) targetEx;
+		}
+		catch (InvocationTargetException e) {
+			if (e.getTargetException() instanceof DataAccessException) {
+				throw (DataAccessException) e.getTargetException();
 			}
 			else {
-				throw new GemfireListenerExecutionFailedException("Listener method '" + methodName
-						+ "' threw exception", targetEx);
+				throw new GemfireListenerExecutionFailedException(
+					String.format("Listener method '%1$s' threw exception...", methodName), e.getTargetException());
 			}
-		} catch (Throwable ex) {
-			throw new GemfireListenerExecutionFailedException("Failed to invoke target method '" + methodName, ex);
+		}
+		catch (Throwable e) {
+			throw new GemfireListenerExecutionFailedException(
+				String.format("Failed to invoke target listener method '%1$s'", methodName), e);
 		}
 	}
+
+	private class MethodInvoker {
+		private final Object delegate;
+		List<Method> methods;
+
+		MethodInvoker(Object delegate, final String methodName) {
+			this.delegate = delegate;
+
+			Class<?> c = delegate.getClass();
+
+			methods = new ArrayList<Method>();
+
+			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) {
+						if (Modifier.isPublic(method.getModifiers()) && methodName.equals(method.getName())) {
+
+							// check out the arguments
+							Class<?>[] parameterTypes = method.getParameterTypes();
+							int objects = 0;
+							int operations = 0;
+
+							if (parameterTypes.length > 0) {
+								for (Class<?> paramType : parameterTypes) {
+
+									if (Object.class.equals(paramType)) {
+										objects++;
+										if (objects > 2) {
+											return false;
+										}
+									}
+									else if (Operation.class.equals(paramType)) {
+										operations++;
+										if (operations > 2) {
+											return false;
+										}
+									}
+									else if (CqEvent.class.equals(paramType)) {
+									}
+									else if (Throwable.class.equals(paramType)) {
+									}
+									else if (byte[].class.equals(paramType)) {
+									}
+									else if (CqQuery.class.equals(paramType)) {
+									}
+									else {
+										return false;
+									}
+								}
+								return true;
+							}
+						}
+						return false;
+					}
+				});
+
+			Assert.isTrue(!methods.isEmpty(), "Cannot find a suitable method named [" + c.getName() + "#" + methodName
+				+ "] - is the method public and has the proper arguments?");
+		}
+
+		void invoke(CqEvent event) throws InvocationTargetException, IllegalAccessException {
+
+			for (Method m : methods) {
+				Class<?>[] types = m.getParameterTypes();
+				Object[] args = new Object[types.length];
+
+				boolean value = false;
+				boolean query = false;
+
+				for (int i = 0; i < types.length; i++) {
+					Class<?> paramType = types[i];
+
+					if (Object.class.equals(paramType)) {
+						args[i] = (!value ? event.getKey() : event.getNewValue());
+						value = true;
+					}
+					else if (Operation.class.equals(paramType)) {
+						args[i] = (!query ? event.getBaseOperation() : event.getQueryOperation());
+						query = true;
+					}
+					else if (CqEvent.class.equals(paramType)) {
+						args[i] = event;
+					}
+					else if (Throwable.class.equals(paramType)) {
+						args[i] = event.getThrowable();
+					}
+					else if (byte[].class.equals(paramType)) {
+						args[i] = event.getDeltaValue();
+					}
+					else if (CqQuery.class.equals(paramType)) {
+						args[i] = event.getCq();
+					}
+				}
+
+				m.invoke(delegate, args);
+			}
+		}
+	}
+
 }

src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.4.xsd

@@ -2097,60 +2097,45 @@ The name of the cache server definition (by default "gemfireServer").
 				]]></xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
-			<xsd:attribute name="auto-startup" type="xsd:string"
-						   use="optional" default="true" />
-			<xsd:attribute name="bind-address" type="xsd:string"
-						   use="optional" />
-			<xsd:attribute name="port" type="xsd:string" use="optional"
-						   default="40404">
+			<xsd:attribute name="cache-ref" type="xsd:string" use="optional" default="gemfireCache">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+The name of the bean defining the GemFire cache (by default 'gemfireCache').
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="auto-startup" type="xsd:string" use="optional" default="true"/>
+			<xsd:attribute name="bind-address" type="xsd:string" use="optional" />
+			<xsd:attribute name="port" type="xsd:string" use="optional" default="40404">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[
 The port number of the server.
 					]]></xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
-			<xsd:attribute name="host-name-for-clients" type="xsd:string"
-						   use="optional" />
-			<xsd:attribute name="load-poll-interval" type="xsd:string"
-						   use="optional" default="5000" />
-			<xsd:attribute name="max-connections" type="xsd:string"
-						   use="optional" default="800" />
-			<xsd:attribute name="max-threads" type="xsd:string"
-						   use="optional" default="0" />
-			<xsd:attribute name="max-message-count" type="xsd:string"
-						   use="optional" default="230000" />
-			<xsd:attribute name="max-time-between-pings" type="xsd:string"
-						   use="optional" default="60000" />
-			<xsd:attribute name="message-time-to-live" type="xsd:string"
-						   use="optional" default="180" />
-			<xsd:attribute name="socket-buffer-size" type="xsd:string"
-						   use="optional" default="32768" />
-			<xsd:attribute name="notify-by-subscription" type="xsd:string"
-						   use="optional" default="true" />
-			<xsd:attribute name="groups" type="xsd:string" use="optional"
-						   default="">
+			<xsd:attribute name="host-name-for-clients" type="xsd:string" use="optional" />
+			<xsd:attribute name="load-poll-interval" type="xsd:string" use="optional" default="5000"/>
+			<xsd:attribute name="max-connections" type="xsd:string" use="optional" default="800"/>
+			<xsd:attribute name="max-threads" type="xsd:string" use="optional" default="0" />
+			<xsd:attribute name="max-message-count" type="xsd:string" use="optional" default="230000"/>
+			<xsd:attribute name="max-time-between-pings" type="xsd:string" use="optional" default="60000"/>
+			<xsd:attribute name="message-time-to-live" type="xsd:string" use="optional" default="180"/>
+			<xsd:attribute name="socket-buffer-size" type="xsd:string" use="optional" default="32768"/>
+			<xsd:attribute name="notify-by-subscription" type="xsd:string" use="optional" default="true"/>
+			<xsd:attribute name="groups" type="xsd:string" use="optional" default="">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[
 The server groups that this server will be a member of given as a comma separated values list.
 				]]></xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
-			<xsd:attribute name="load-probe-ref" type="xsd:string"
-						   use="optional">
+			<xsd:attribute name="load-probe-ref" type="xsd:string" use="optional">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[
 The name of the bean defining the CacheServer Load Probe.
 					]]></xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
-			<xsd:attribute name="cache-ref" type="xsd:string" use="optional"
-						   default="gemfireCache">
-				<xsd:annotation>
-					<xsd:documentation><![CDATA[
-The name of the bean defining the GemFire cache (by default 'gemfireCache').
-					]]></xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
 		</xsd:complexType>
 	</xsd:element>
 	<!-- -->
@@ -2168,20 +2153,19 @@ Container for continuous query listeners. All listeners will be hosted by the sa
 		</xsd:annotation>
 		<xsd:complexType>
 			<xsd:sequence>
-				<xsd:element name="listener" type="listenerType"
-							 minOccurs="0" maxOccurs="unbounded" />
+				<xsd:element name="listener" type="listenerType" minOccurs="0" maxOccurs="unbounded"/>
 			</xsd:sequence>
 			<xsd:attribute name="id" type="xsd:string" use="optional">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[
 The id of the listener (optional)
-                                    ]]></xsd:documentation>
+					]]></xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
 			<xsd:attribute name="cache" type="xsd:string" default="gemfireCache">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[
-A reference (by name) to the GemFire cache bean. Default is "gemfireCache".
+A reference (by name) to the GemFire Cache bean. Default is "gemfireCache".
 	          ]]></xsd:documentation>
 					<xsd:appinfo>
 						<tool:annotation kind="ref">
@@ -2190,19 +2174,14 @@ A reference (by name) to the GemFire cache bean. Default is "gemfireCache".
 					</xsd:appinfo>
 				</xsd:annotation>
 			</xsd:attribute>
-			<xsd:attribute name="task-executor" type="xsd:string">
+			<xsd:attribute name="pool-name" type="xsd:string" use="optional">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[
-A reference to a Spring TaskExecutor (or standard JDK 1.5 Executor) for executing
-GemFire listener invokers. Default is a SimpleAsyncTaskExecutor.
-	          ]]></xsd:documentation>
-					<xsd:appinfo>
-						<tool:annotation kind="ref">
-							<tool:expected-type type="java.util.concurrent.Executor" />
-						</tool:annotation>
-					</xsd:appinfo>
+The name of the GemFire Pool used by the container.
+ 						]]></xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="auto-startup" type="xsd:string" use="optional" default="true"/>
 			<xsd:attribute name="phase" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[
@@ -2213,11 +2192,29 @@ and stop as soon as possible.
 	          ]]></xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
-			<xsd:attribute name="pool-name" type="xsd:string" use="optional">
+			<xsd:attribute name="error-handler" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[
-The name of the pool used by the container.
- 						]]></xsd:documentation>
+A reference to a Spring ErrorHandler strategy handling any errors that may occur when the container executes the CQs.
+	          		]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="java.util.concurrent.Executor"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="task-executor" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+A reference to a Spring TaskExecutor (or standard JDK 1.5 Executor) for executing GemFire CQ listener invokers.
+The default is a SimpleAsyncTaskExecutor.
+	          		]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="java.util.concurrent.Executor"/>
+						</tool:annotation>
+					</xsd:appinfo>
 				</xsd:annotation>
 			</xsd:attribute>
 		</xsd:complexType>
@@ -2239,6 +2236,13 @@ Required.
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
 The query for the GemFire continuous query.
+	        ]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="name" type="xsd:string" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The name of the resulting GemFire ContinuousQuery (CQ). Useful for monitoring and statistics-based querying.
 	        ]]></xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
@@ -2247,13 +2251,6 @@ The query for the GemFire continuous query.
 				<xsd:documentation><![CDATA[
 The name of the listener method to invoke. If not specified, the target bean is supposed to implement the ContinuousQueryListener
 interface or provide a method named 'handleEvent'.
-	        ]]></xsd:documentation>
-			</xsd:annotation>
-		</xsd:attribute>
-		<xsd:attribute name="name" type="xsd:string" use="optional">
-			<xsd:annotation>
-				<xsd:documentation><![CDATA[
-The name of the resulting GemFire continuous query. Useful for monitoring and statistics querying.
 	        ]]></xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>

src/test/java/org/springframework/data/gemfire/config/ContinuousQueryListenerContainerNamespaceTest.java

@@ -0,0 +1,137 @@
+/*
+ * Copyright 2010-2013 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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Executor;
+import javax.annotation.Resource;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.gemfire.ForkUtil;
+import org.springframework.data.gemfire.TestUtils;
+import org.springframework.data.gemfire.listener.ContinuousQueryListener;
+import org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer;
+import org.springframework.data.gemfire.listener.GemfireMDP;
+import org.springframework.data.gemfire.listener.adapter.ContinuousQueryListenerAdapter;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.util.ErrorHandler;
+
+import com.gemstone.gemfire.cache.client.ClientCache;
+import com.gemstone.gemfire.cache.query.CqListener;
+import com.gemstone.gemfire.cache.query.CqQuery;
+
+/**
+ * The ContinuousQueryListenerContainerNamespaceTest class is a test suite of test cases testing the SDG XML namespace
+ * for proper configuration and initialization of a ContinuousQueryListenerContainer bean component
+ * in the Spring context.
+ *
+ * @author John Blum
+ * @see org.junit.Test
+ * @see org.junit.runner.RunWith
+ * @see org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer
+ * @see org.springframework.test.context.ContextConfiguration
+ * @see org.springframework.test.context.junit4.SpringJUnit4ClassRunner
+ * @see com.gemstone.gemfire.cache.client.ClientCache
+ * @see com.gemstone.gemfire.cache.query.CqListener
+ * @see com.gemstone.gemfire.cache.query.CqQuery
+ * @since 1.4.0
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@SuppressWarnings("unused")
+public class ContinuousQueryListenerContainerNamespaceTest {
+
+	@BeforeClass
+	public static void setupBeforeClass() {
+		ForkUtil.cacheServer();
+	}
+
+	@AfterClass
+	public static void tearDownAfterClass() {
+		ForkUtil.sendSignal();
+	}
+
+	@Autowired
+	private ClientCache gemfireCache;
+
+	@Autowired
+	private ContinuousQueryListenerContainer container;
+
+	@Resource(name = "testErrorHandler")
+	private ErrorHandler testErrorHandler;
+
+	@Resource(name = "testTaskExecutor")
+	private Executor testTaskExecutor;
+
+	@Test
+	public void testContainerConfiguration() throws Exception {
+		assertNotNull("The ContinuousQueryListenerContainer was not properly configured!", container);
+		assertTrue("The CQ Listener Container should be active (initialized)!", container.isActive());
+		assertFalse("The CQ Listener container should not be configured to auto-start!", container.isAutoStartup());
+		assertFalse("The CQ Listener Container should not be running!", container.isRunning());
+		assertEquals(4, container.getPhase());
+		assertNotNull(testErrorHandler);
+		assertSame(testErrorHandler, TestUtils.readField("errorHandler", container));
+		assertNotNull(testTaskExecutor);
+		assertSame(testTaskExecutor, TestUtils.readField("taskExecutor", container));
+
+		CqQuery[] queries = gemfireCache.getQueryService().getCqs();
+
+		assertNotNull(queries);
+		assertEquals(3, queries.length);
+
+		List<String> actualNames = new ArrayList<String>(3);
+
+		for (CqQuery query : queries) {
+			actualNames.add(query.getName());
+			assertEquals("SELECT * FROM /test-cq", query.getQueryString());
+			assertEquals("Q3".equalsIgnoreCase(query.getName()), query.isDurable());
+
+			CqListener cqListener = query.getCqAttributes().getCqListener();
+
+			assertNotNull(cqListener);
+
+			// the CqListener object should be an instance of...
+			// org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer.EventDispatcherAdapter
+			// So, get the SDG "ContinuousQueryListener"...
+			ContinuousQueryListener listener = TestUtils.readField("delegate", cqListener);
+
+			assertTrue(listener instanceof ContinuousQueryListenerAdapter);
+			assertTrue(((ContinuousQueryListenerAdapter) listener).getDelegate() instanceof GemfireMDP);
+
+			if ("Q2".equalsIgnoreCase(query.getName())) {
+				assertEquals("handleQuery", TestUtils.readField("defaultListenerMethod", listener));
+			}
+		}
+
+		actualNames.containsAll(Arrays.asList("Q1", "Q2", "Q3"));
+	}
+
+}

src/test/java/org/springframework/data/gemfire/fork/CacheServerProcess.java

@@ -32,45 +32,46 @@ import com.gemstone.gemfire.cache.server.CacheServer;
 
 /**
  * @author Costin Leau
+ * @author John Blum
  */
 @SuppressWarnings("unchecked")
 public class CacheServerProcess {
 
-	public static void main(String[] args) throws Exception {
-
+	public static void main(final String[] args) throws Exception {
 		Properties props = new Properties();
 		props.setProperty("name", "CqServer");
+		props.setProperty("mcast-port", "0");
 		props.setProperty("log-level", "warning");
 
 		Cache cache = new CacheFactory(props).create();
 
-		// Create region.
-		// Create region.
-		RegionFactory<Object,Object> factory = cache.createRegionFactory();
-		factory.setDataPolicy(DataPolicy.REPLICATE);
-		factory.setScope(Scope.DISTRIBUTED_ACK);
-		Region testRegion = factory.create("test-cq");
-		System.out.println("Test region, " + testRegion.getFullPath() + ", created in cache.");
-	 
-		// Start Cache Server.
+		RegionFactory<String, Integer> regionFactory = cache.createRegionFactory();
+		regionFactory.setDataPolicy(DataPolicy.REPLICATE);
+		regionFactory.setScope(Scope.DISTRIBUTED_ACK);
+
+		Region<String, Integer> testRegion = regionFactory.create("test-cq");
+
+		System.out.printf("Test Region '%1$s' created in Cache '%2$s.%n", testRegion.getFullPath(), cache.getName());
+
 		CacheServer server = cache.addCacheServer();
 		server.setPort(40404);
 		server.start();
 
 		ForkUtil.createControlFile(CacheServerProcess.class.getName());
-		
-		System.out.println("Waiting for signal");
-		// wait for signal
+
+		System.out.println("Waiting for signal...");
+
 		BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in));
 		bufferedReader.readLine();
 
-		System.out.println("Received signal");
+		System.out.println("Signal received!");
 
 		testRegion.put("one", 1);
 		testRegion.put("two", 2);
 		testRegion.put("three", 3);
 
-		System.out.println("Waiting for shutdown");
+		System.out.println("Waiting for shutdown...");
+
 		bufferedReader.readLine();
 	}
 

src/test/java/org/springframework/data/gemfire/listener/adapter/ContainerXmlSetupTest.java

@@ -17,6 +17,8 @@
 package org.springframework.data.gemfire.listener.adapter;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 
 import org.junit.AfterClass;
@@ -24,8 +26,6 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 import org.springframework.context.support.GenericXmlApplicationContext;
 import org.springframework.data.gemfire.ForkUtil;
-import org.springframework.data.gemfire.listener.ContinuousQueryDefinition;
-import org.springframework.data.gemfire.listener.ContinuousQueryListener;
 import org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer;
 
 import com.gemstone.gemfire.cache.Cache;
@@ -34,6 +34,7 @@ import com.gemstone.gemfire.cache.query.CqQuery;
 
 /**
  * @author Costin Leau
+ * @author John Blum
  */
 public class ContainerXmlSetupTest {
 
@@ -47,28 +48,38 @@ public class ContainerXmlSetupTest {
 		ForkUtil.sendSignal();
 	}
 
-
 	@Test
 	public void testContainerSetup() throws Exception {
-		GenericXmlApplicationContext ctx = new GenericXmlApplicationContext(
-				"/org/springframework/data/gemfire/listener/container.xml");
+		GenericXmlApplicationContext applicationContext = new GenericXmlApplicationContext(
+			"/org/springframework/data/gemfire/listener/container.xml");
 
-		ContinuousQueryListenerContainer container = ctx.getBean(ContinuousQueryListenerContainer.class);
-		assertTrue(container.isRunning());
+		try {
+			ContinuousQueryListenerContainer container = applicationContext.getBean(
+				ContinuousQueryListenerContainer.class);
 
-		Cache cache = ctx.getBean("gemfireCache", Cache.class);
-		Pool pool = ctx.getBean("client", Pool.class);
-		
-		// Test getting container listener bean by ID
-		ctx.getBean("testContainerId");
+			assertNotNull(container);
+			assertTrue(container.isRunning());
 
-		CqQuery[] cqs = cache.getQueryService().getCqs();
-		CqQuery[] pcqs = pool.getQueryService().getCqs();
-		assertTrue(pool.getQueryService().getCq("test-bean-1") != null);
-		assertEquals(3, cqs.length);
-		assertEquals(3, pcqs.length);
-		ForkUtil.sendSignal();
-		ctx.close();
-		
+			// test getting container listener by bean ID
+			ContinuousQueryListenerContainer container2 = applicationContext.getBean("testContainerId",
+				ContinuousQueryListenerContainer.class);
+
+			assertSame(container, container2);
+
+			Cache cache = applicationContext.getBean("gemfireCache", Cache.class);
+			Pool pool = applicationContext.getBean("client", Pool.class);
+
+			CqQuery[] cqs = cache.getQueryService().getCqs();
+			CqQuery[] poolCqs = pool.getQueryService().getCqs();
+
+			assertTrue(pool.getQueryService().getCq("test-bean-1") != null);
+			assertEquals(3, cqs.length);
+			assertEquals(3, poolCqs.length);
+		}
+		finally {
+			ForkUtil.sendSignal();
+			applicationContext.close();
+		}
 	}
+
 }

src/test/resources/org/springframework/data/gemfire/config/ContinuousQueryListenerContainerNamespaceTest-context.xml

@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:gfe="http://www.springframework.org/schema/gemfire"
+	   xmlns:task="http://www.springframework.org/schema/task"
+	   xmlns:util="http://www.springframework.org/schema/util"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xsi:schemaLocation="
+	   	http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+	   	http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd
+	   	http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
+	   	http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
+">
+
+	<util:properties id="clientCacheConfigurationSettings">
+		<prop key="name">cq-client</prop>
+		<prop key="log-level">warning</prop>
+	</util:properties>
+
+	<gfe:pool id="clientPool" subscription-enabled="true">
+		<gfe:server host="localhost" port="40404"/>
+	</gfe:pool>
+
+	<gfe:client-cache use-bean-factory-locator="false" properties-ref="clientCacheConfigurationSettings" pool-name="clientPool"/>
+
+	<bean id="testErrorHandler" class="org.springframework.data.gemfire.listener.StubErrorHandler"/>
+	<bean id="testQueryListener" class="org.springframework.data.gemfire.listener.GemfireMDP"/>
+
+	<task:executor id="testTaskExecutor"/>
+
+	<gfe:cq-listener-container id="testContainerId" cache="gemfireCache" pool-name="clientPool"
+							   auto-startup="false" phase="4"
+							   error-handler="testErrorHandler" task-executor="testTaskExecutor">
+		<gfe:listener ref="testQueryListener" query="SELECT * FROM /test-cq" name="Q1"/>
+		<gfe:listener ref="testQueryListener" query="SELECT * FROM /test-cq" name="Q2" method="handleQuery"/>
+		<gfe:listener ref="testQueryListener" query="SELECT * FROM /test-cq" name="Q3" durable="true"/>
+	</gfe:cq-listener-container>
+
+</beans>