+ rename Cq to ContinousQuery
+ several improvements
+ namespace for cq-listener-container
This commit is contained in:
Costin Leau
2011-08-24 19:13:26 +03:00
parent b12a34982a
commit 865c0676d3
16 changed files with 522 additions and 73 deletions

View File

@@ -0,0 +1,126 @@
/*
* Copyright 2011 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 java.util.List;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.ManagedSet;
import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.data.gemfire.listener.ContinousQueryDefinition;
import org.springframework.data.gemfire.listener.ContinousQueryListenerContainer;
import org.springframework.data.gemfire.listener.adapter.ContinousQueryListenerAdapter;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Attr;
import org.w3c.dom.Element;
import org.w3c.dom.NamedNodeMap;
/**
* Parser for SGF <code>&lt;cq-listener-container&gt;</code> element.
*
* @author Costin Leau
*/
class GemfireListenerContainerParser extends AbstractSimpleBeanDefinitionParser {
@Override
protected Class<ContinousQueryListenerContainer> getBeanClass(Element element) {
return ContinousQueryListenerContainer.class;
}
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
// parse attributes (but replace the value assignment with references)
NamedNodeMap attributes = element.getAttributes();
for (int x = 0; x < attributes.getLength(); x++) {
Attr attribute = (Attr) attributes.item(x);
if (isEligibleAttribute(attribute, parserContext)) {
String propertyName = extractPropertyName(attribute.getLocalName());
Assert.state(StringUtils.hasText(propertyName),
"Illegal property name returned from 'extractPropertyName(String)': cannot be null or empty.");
builder.addPropertyReference(propertyName, attribute.getValue());
}
}
String phase = element.getAttribute("phase");
if (StringUtils.hasText(phase)) {
builder.addPropertyValue("phase", phase);
}
postProcess(builder, element);
// parse nested listeners
List<Element> listDefs = DomUtils.getChildElementsByTagName(element, "listener");
if (!listDefs.isEmpty()) {
ManagedSet<BeanDefinition> listeners = new ManagedSet<BeanDefinition>(listDefs.size());
for (Element listElement : listDefs) {
listeners.add(parseListener(listElement));
}
builder.addPropertyValue("queryListeners", listeners);
}
}
@Override
protected boolean isEligibleAttribute(String attributeName) {
return (!"phase".equals(attributeName));
}
/**
* Parses a listener definition. Returns the listener bean reference definition (of a {@link ContinousQueryDefinition}).
*
* @param element
* @return
*/
private BeanDefinition parseListener(Element element) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ContinousQueryListenerAdapter.class);
builder.addConstructorArgReference(element.getAttribute("ref"));
String attr = element.getAttribute("method");
if (StringUtils.hasText(attr)) {
builder.addPropertyValue("defaultListenerMethod", attr);
}
BeanDefinitionBuilder defBuilder = BeanDefinitionBuilder.genericBeanDefinition(ContinousQueryDefinition.class);
attr = element.getAttribute("name");
if (StringUtils.hasText(attr)) {
defBuilder.addConstructorArgValue(attr);
}
defBuilder.addConstructorArgValue(element.getAttribute("query"));
defBuilder.addConstructorArgValue(builder.getBeanDefinition());
attr = element.getAttribute("durable");
if (StringUtils.hasText(attr)) {
defBuilder.addConstructorArgValue(attr);
}
return defBuilder.getBeanDefinition();
}
@Override
protected boolean shouldGenerateId() {
return true;
}
}

View File

@@ -35,5 +35,7 @@ class GemfireNamespaceHandler extends NamespaceHandlerSupport {
registerBeanDefinitionParser("cache-server", new CacheServerParser());
registerBeanDefinitionParser("transaction-manager", new TransactionManagerParser());
registerBeanDefinitionParser("cq-listener-container", new GemfireListenerContainerParser());
}
}

View File

@@ -22,36 +22,36 @@ import org.springframework.util.Assert;
import com.gemstone.gemfire.cache.query.CqQuery;
/**
* Basic holder class for defining an CqQuery. Useful for configuring GemFire {@link CqQuery}s through XML
* Basic holder class for defining an {@link CqQuery}. Useful for configuring GemFire {@link CqQuery}s through XML
* and or JavaBeans means.
*
* @author Costin Leau
*/
public class CqQueryDefinition implements InitializingBean {
public class ContinousQueryDefinition implements InitializingBean {
private String name = null, query = null;
private QueryListener listener = null;
private ContinuousQueryListener listener = null;
private boolean durable = false;
public CqQueryDefinition() {
public ContinousQueryDefinition() {
}
public CqQueryDefinition(String query, QueryListener listener) {
public ContinousQueryDefinition(String query, ContinuousQueryListener listener) {
this(query, listener, false);
}
public CqQueryDefinition(String query, QueryListener listener, boolean durable) {
public ContinousQueryDefinition(String query, ContinuousQueryListener listener, boolean durable) {
this.query = query;
this.listener = listener;
this.durable = durable;
afterPropertiesSet();
}
public CqQueryDefinition(String name, String query, QueryListener listener) {
public ContinousQueryDefinition(String name, String query, ContinuousQueryListener listener) {
this(name, query, listener, false);
}
public CqQueryDefinition(String name, String query, QueryListener listener, boolean durable) {
public ContinousQueryDefinition(String name, String query, ContinuousQueryListener listener, boolean durable) {
this.name = name;
this.query = query;
this.listener = listener;
@@ -81,7 +81,7 @@ public class CqQueryDefinition implements InitializingBean {
/**
* @return the listener
*/
public QueryListener getListener() {
public ContinuousQueryListener getListener() {
return listener;
}

View File

@@ -47,12 +47,12 @@ import com.gemstone.gemfire.internal.concurrent.ConcurrentHashSet;
*
* @author Costin Leau
*/
public class QueryListenerContainer implements InitializingBean, DisposableBean, BeanNameAware, SmartLifecycle {
public class ContinousQueryListenerContainer implements InitializingBean, DisposableBean, BeanNameAware, SmartLifecycle {
private class EventDispatcherAdapter implements CqListener {
private final QueryListener delegate;
private final ContinuousQueryListener delegate;
EventDispatcherAdapter(QueryListener delegate) {
EventDispatcherAdapter(ContinuousQueryListener delegate) {
this.delegate = delegate;
}
@@ -72,9 +72,9 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean,
protected final Log logger = LogFactory.getLog(getClass());
/**
* Default thread name prefix: "QueryListenerContainer-".
* Default thread name prefix: "ContinousQueryListenerContainer-".
*/
public static final String DEFAULT_THREAD_NAME_PREFIX = ClassUtils.getShortName(QueryListenerContainer.class) + "-";
public static final String DEFAULT_THREAD_NAME_PREFIX = ClassUtils.getShortName(ContinousQueryListenerContainer.class) + "-";
private Executor subscriptionExecutor;
private Executor taskExecutor;
@@ -160,7 +160,7 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean,
doStart();
if (logger.isDebugEnabled()) {
logger.debug("Started QueryListenerContainer");
logger.debug("Started ContinousQueryListenerContainer");
}
}
}
@@ -172,7 +172,7 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean,
}
if (logger.isDebugEnabled()) {
logger.debug("Stopped QueryListenerContainer");
logger.debug("Stopped ContinousQueryListenerContainer");
}
}
@@ -215,7 +215,7 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean,
*
* @see #handleListenerException
*/
protected void executeListener(QueryListener listener, CqEvent event) {
protected void executeListener(ContinuousQueryListener listener, CqEvent event) {
try {
listener.onEvent(event);
} catch (Throwable ex) {
@@ -298,6 +298,11 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean,
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;
}
@@ -307,7 +312,7 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean,
*
* @param queries set of queries
*/
public void setQueryListeners(Set<CqQueryDefinition> queries) {
public void setQueryListeners(Set<ContinousQueryDefinition> queries) {
initMapping(queries);
}
@@ -317,11 +322,11 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean,
*
* @param listener event cqQuery
*/
public void addListener(CqQueryDefinition cqQuery) {
public void addListener(ContinousQueryDefinition cqQuery) {
doAddListener(cqQuery);
}
private void initMapping(Set<CqQueryDefinition> queryDefinitions) {
private void initMapping(Set<ContinousQueryDefinition> queryDefinitions) {
// stop the listener if currently running
if (isRunning()) {
stop();
@@ -329,7 +334,7 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean,
closeQueries();
for (CqQueryDefinition def : queryDefinitions) {
for (ContinousQueryDefinition def : queryDefinitions) {
doAddListener(def);
}
@@ -339,7 +344,7 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean,
}
}
private void doAddListener(CqQueryDefinition def) {
private void doAddListener(ContinousQueryDefinition def) {
CqQuery cq = null;
try {
@@ -375,7 +380,7 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean,
}
}
private void dispatchEvent(final QueryListener listener, final CqEvent event) {
private void dispatchEvent(final ContinuousQueryListener listener, final CqEvent event) {
taskExecutor.execute(new Runnable() {
public void run() {
executeListener(listener, event);

View File

@@ -23,7 +23,7 @@ import com.gemstone.gemfire.cache.query.CqEvent;
*
* @author Costin Leau
*/
public interface QueryListener {
public interface ContinuousQueryListener {
void onEvent(CqEvent event);
}

View File

@@ -17,13 +17,13 @@
package org.springframework.data.gemfire.listener;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.gemfire.listener.adapter.QueryListenerAdapter;
import org.springframework.data.gemfire.listener.adapter.ContinousQueryListenerAdapter;
/**
* Exception thrown when the execution of a listener method failed.
*
* @author Costin Leau
* @see QueryListenerAdapter
* @see ContinousQueryListenerAdapter
*/
public class GemfireListenerExecutionFailedException extends InvalidDataAccessApiUsageException {

View File

@@ -27,7 +27,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.gemfire.listener.GemfireListenerExecutionFailedException;
import org.springframework.data.gemfire.listener.QueryListener;
import org.springframework.data.gemfire.listener.ContinuousQueryListener;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.ReflectionUtils.MethodCallback;
@@ -69,7 +69,7 @@ import com.gemstone.gemfire.cache.query.CqQuery;
* @author Costin Leau
* @see org.springframework.jms.listener.adapter.MessageListenerAdapter
*/
public class QueryListenerAdapter implements QueryListener {
public class ContinousQueryListenerAdapter implements ContinuousQueryListener {
private class MethodInvoker {
private final Object delegate;
@@ -192,18 +192,18 @@ public class QueryListenerAdapter implements QueryListener {
private MethodInvoker invoker;
/**
* Create a new {@link QueryListenerAdapter} with default settings.
* Create a new {@link ContinousQueryListenerAdapter} with default settings.
*/
public QueryListenerAdapter() {
public ContinousQueryListenerAdapter() {
setDelegate(this);
}
/**
* Create a new {@link QueryListenerAdapter} for the given delegate.
* Create a new {@link ContinousQueryListenerAdapter} for the given delegate.
*
* @param delegate the delegate object
*/
public QueryListenerAdapter(Object delegate) {
public ContinousQueryListenerAdapter(Object delegate) {
setDelegate(delegate);
}
@@ -251,7 +251,7 @@ public class QueryListenerAdapter implements QueryListener {
}
/**
* Standard {@link QueryListener} entry point.
* Standard {@link ContinuousQueryListener} entry point.
* <p>Delegates the event to the target listener method, with appropriate
* conversion of the event argument. In case of an exception, the
* {@link #handleListenerException(Throwable)} method will be invoked.
@@ -263,11 +263,11 @@ public class QueryListenerAdapter implements QueryListener {
public void onEvent(CqEvent event) {
try {
// Check whether the delegate is a QueryListener impl itself.
// Check whether the delegate is a ContinuousQueryListener impl itself.
// In that case, the adapter will simply act as a pass-through.
if (delegate != this) {
if (delegate instanceof QueryListener) {
((QueryListener) delegate).onEvent(event);
if (delegate instanceof ContinuousQueryListener) {
((ContinuousQueryListener) delegate).onEvent(event);
}
}