Spring/spring-data-geode
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fixes JIRA issues SGF-271 - <cq-listener-container> element 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.

Browse Source
This commit is contained in:
John Blum
2014-04-18 23:01:02 -07:00
parent 3e300ddaf5
commit 163e36497c
10 changed files with 921 additions and 632 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
src/main/java/org/springframework/data/gemfire/config/GemfireListenerContainerParser.java
View File

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

47
src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryDefinition.java
View File

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

10
src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListener.java
View File

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

776
src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java
View File

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

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

@@ -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.
*
* <p/>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,130 +69,21 @@ import com.gemstone.gemfire.cache.query.CqQuery;
* @author Juergen Hoeller
* @author Costin Leau
* @author Oliver Gierke
* @see org.springframework.jms.listener.adapter.MessageListenerAdapter
* @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.
*/
@@ -209,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.
@@ -220,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;
}
@@ -263,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) {
@@ -273,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) {
@@ -290,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.
@@ -312,6 +194,15 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
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
@@ -321,17 +212,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);
}
}
}
}