Invoked after population of normal bean properties but before an
+ * init callback such as {@link InitializingBean#afterPropertiesSet()}
+ * or a custom init-method.
+ *
+ * @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 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.
+ * 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 defs = new LinkedHashSet();
- private Set queries = new ConcurrentHashSet();
-
- 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.
- * 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.
- *
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 no 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 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 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);
- }
- });
- }
-}
\ No newline at end of file
+}
diff --git a/src/main/java/org/springframework/data/gemfire/listener/adapter/ContinuousQueryListenerAdapter.java b/src/main/java/org/springframework/data/gemfire/listener/adapter/ContinuousQueryListenerAdapter.java
index 2ee8ade8..36c6649e 100644
--- a/src/main/java/org/springframework/data/gemfire/listener/adapter/ContinuousQueryListenerAdapter.java
+++ b/src/main/java/org/springframework/data/gemfire/listener/adapter/ContinuousQueryListenerAdapter.java
@@ -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
+ * Modeled as much as possible after the JMS MessageListenerAdapter in
* Spring Framework.
*
*
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 methods;
-
- MethodInvoker(Object delegate, final String methodName) {
- this.delegate = delegate;
-
- Class c = delegate.getClass();
-
- methods = new ArrayList();
-
- ReflectionUtils.doWithMethods(c, new MethodCallback() {
-
- public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
- ReflectionUtils.makeAccessible(method);
- methods.add(method);
- }
-
- }, new MethodFilter() {
- public boolean matches(Method method) {
- 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 null)
* @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 methods;
+
+ MethodInvoker(Object delegate, final String methodName) {
+ this.delegate = delegate;
+
+ Class c = delegate.getClass();
+
+ methods = new ArrayList();
+
+ ReflectionUtils.doWithMethods(c, new MethodCallback() {
+
+ public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
+ ReflectionUtils.makeAccessible(method);
+ methods.add(method);
+ }
+
+ }, new MethodFilter() {
+ public boolean matches(Method method) {
+ 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);
+ }
+ }
+ }
+
}
diff --git a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.4.xsd b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.4.xsd
index bf3ab5c5..4853e9eb 100644
--- a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.4.xsd
+++ b/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").
]]>
-
-
-
+
+
+
+
+
+
+
+
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
-
+
-
-
-
-
-
@@ -2168,20 +2153,19 @@ Container for continuous query listeners. All listeners will be hosted by the sa
-
+
+ ]]>
@@ -2190,19 +2174,14 @@ A reference (by name) to the GemFire cache bean. Default is "gemfireCache".
-
+
-
-
-
-
-
+The name of the GemFire Pool used by the container.
+ ]]>
+
-
+
+A reference to a Spring ErrorHandler strategy handling any errors that may occur when the container executes the CQs.
+ ]]>
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
@@ -2239,6 +2236,13 @@ Required.
+
+
+
+
+
@@ -2247,13 +2251,6 @@ The query for the GemFire continuous query.
-
-
-
-
-
diff --git a/src/test/java/org/springframework/data/gemfire/config/ContinuousQueryListenerContainerNamespaceTest.java b/src/test/java/org/springframework/data/gemfire/config/ContinuousQueryListenerContainerNamespaceTest.java
new file mode 100644
index 00000000..c3cf6705
--- /dev/null
+++ b/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 actualNames = new ArrayList(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"));
+ }
+
+}
diff --git a/src/test/java/org/springframework/data/gemfire/fork/CacheServerProcess.java b/src/test/java/org/springframework/data/gemfire/fork/CacheServerProcess.java
index b1434914..e2f29b12 100644
--- a/src/test/java/org/springframework/data/gemfire/fork/CacheServerProcess.java
+++ b/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 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 regionFactory = cache.createRegionFactory();
+ regionFactory.setDataPolicy(DataPolicy.REPLICATE);
+ regionFactory.setScope(Scope.DISTRIBUTED_ACK);
+
+ Region 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();
}
diff --git a/src/test/java/org/springframework/data/gemfire/listener/adapter/ContainerXmlSetupTest.java b/src/test/java/org/springframework/data/gemfire/listener/adapter/ContainerXmlSetupTest.java
index e8e9bde3..02583be2 100644
--- a/src/test/java/org/springframework/data/gemfire/listener/adapter/ContainerXmlSetupTest.java
+++ b/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();
+ }
}
+
}
diff --git a/src/test/resources/org/springframework/data/gemfire/config/ContinuousQueryListenerContainerNamespaceTest-context.xml b/src/test/resources/org/springframework/data/gemfire/config/ContinuousQueryListenerContainerNamespaceTest-context.xml
new file mode 100644
index 00000000..c8923abe
--- /dev/null
+++ b/src/test/resources/org/springframework/data/gemfire/config/ContinuousQueryListenerContainerNamespaceTest-context.xml
@@ -0,0 +1,38 @@
+
+
+
+
+ cq-client
+ warning
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+