SGF-416 - Add additional unit and integration tests.
This commit is contained in:
@@ -28,10 +28,12 @@ import com.gemstone.gemfire.cache.client.Pool;
|
||||
import com.gemstone.gemfire.cache.query.QueryService;
|
||||
|
||||
/**
|
||||
* The DefaultableDelegatingPoolAdapter class...
|
||||
* The DefaultableDelegatingPoolAdapter class is a wrapper class around Pool allowing default configuration property
|
||||
* values to be providing in the case that the Pool's setting were null.
|
||||
*
|
||||
* @author John Blum
|
||||
* @since 1.0.0
|
||||
* @see com.gemstone.gemfire.cache.client.Pool
|
||||
* @since 1.8.0
|
||||
*/
|
||||
@SuppressWarnings("unused")
|
||||
public abstract class DefaultableDelegatingPoolAdapter {
|
||||
@@ -201,8 +203,12 @@ public abstract class DefaultableDelegatingPoolAdapter {
|
||||
}
|
||||
|
||||
/* (non-Javadoc) */
|
||||
public QueryService getQueryService() {
|
||||
return getDelegate().getQueryService();
|
||||
public QueryService getQueryService(QueryService defaultQueryService) {
|
||||
return defaultIfNull(defaultQueryService, new ValueProvider<QueryService>() {
|
||||
@Override public QueryService getValue() {
|
||||
return getDelegate().getQueryService();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/* (non-Javadoc) */
|
||||
|
||||
@@ -116,7 +116,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
|
||||
}
|
||||
|
||||
/* (non-Javadoc) */
|
||||
private String resolvePoolName() {
|
||||
String resolvePoolName() {
|
||||
String poolName = this.poolName;
|
||||
|
||||
if (!StringUtils.hasText(poolName)) {
|
||||
@@ -129,29 +129,37 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
|
||||
}
|
||||
|
||||
/* (non-Javadoc) */
|
||||
private String eagerlyInitializePool(String poolName) {
|
||||
String eagerlyInitializePool(String poolName) {
|
||||
try {
|
||||
if (beanFactory != null && beanFactory.isTypeMatch(poolName, Pool.class)) {
|
||||
beanFactory.getBean(poolName, Pool.class);
|
||||
}
|
||||
}
|
||||
catch (BeansException ignore) {
|
||||
Assert.notNull(PoolManager.find(poolName), String.format("No GemFire Pool with name [%1$s] was found.",
|
||||
Assert.notNull(PoolManager.find(poolName), String.format("No GemFire Pool with name [%1$s] was found",
|
||||
poolName));
|
||||
}
|
||||
|
||||
return poolName;
|
||||
}
|
||||
|
||||
private void initQueryService(String poolName) {
|
||||
queryService = PoolManager.find(poolName).getQueryService();
|
||||
/* (non-Javadoc) */
|
||||
QueryService initQueryService(String poolName) {
|
||||
if (queryService == null) {
|
||||
queryService = PoolManager.find(poolName).getQueryService();
|
||||
}
|
||||
|
||||
return queryService;
|
||||
}
|
||||
|
||||
private void initExecutor() {
|
||||
/* (non-Javadoc) */
|
||||
Executor initExecutor() {
|
||||
if (taskExecutor == null) {
|
||||
manageExecutor = true;
|
||||
taskExecutor = createDefaultTaskExecutor();
|
||||
manageExecutor = true;
|
||||
}
|
||||
|
||||
return taskExecutor;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -204,11 +212,11 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
|
||||
cq.execute();
|
||||
}
|
||||
catch (QueryException ex) {
|
||||
throw new GemfireQueryException(String.format("Could not execute query '%1$s'; state is '%2$s'.",
|
||||
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'.",
|
||||
throw new GemfireQueryException(String.format("Could not execute query [%1$s]; state is [%2$s].",
|
||||
cq.getName(), cq.getState()), ex);
|
||||
}
|
||||
}
|
||||
@@ -286,12 +294,12 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
|
||||
}
|
||||
|
||||
/**
|
||||
* Determines whether the container has be started and is currently running.
|
||||
* Sets whether the CQ listener container should automatically start on startup.
|
||||
*
|
||||
* @return a boolean value indicating whether the container has been started and is currently running.
|
||||
* @param autoStartup a boolean value indicating whether this CQ listener container should automatically start.
|
||||
*/
|
||||
public synchronized boolean isRunning() {
|
||||
return running;
|
||||
public void setAutoStartup(final boolean autoStartup) {
|
||||
this.autoStartup = autoStartup;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -305,12 +313,12 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets whether the CQ listener container should automatically start on startup.
|
||||
* Determines whether the container has be started and is currently running.
|
||||
*
|
||||
* @param autoStartup a boolean value indicating whether this CQ listener container should automatically start.
|
||||
* @return a boolean value indicating whether the container has been started and is currently running.
|
||||
*/
|
||||
public void setAutoStartup(final boolean autoStartup) {
|
||||
this.autoStartup = autoStartup;
|
||||
public synchronized boolean isRunning() {
|
||||
return running;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -357,6 +365,15 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
|
||||
this.errorHandler = errorHandler;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the phase in which this CQ listener container will start in the Spring container.
|
||||
*
|
||||
@@ -367,15 +384,6 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
|
||||
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.
|
||||
*
|
||||
|
||||
@@ -25,7 +25,7 @@ import org.springframework.data.gemfire.listener.adapter.ContinuousQueryListener
|
||||
* @author Costin Leau
|
||||
* @see ContinuousQueryListenerAdapter
|
||||
*/
|
||||
@SuppressWarnings("serial")
|
||||
@SuppressWarnings({ "serial", "unused" })
|
||||
public class GemfireListenerExecutionFailedException extends InvalidDataAccessApiUsageException {
|
||||
|
||||
/**
|
||||
|
||||
@@ -74,7 +74,7 @@ import com.gemstone.gemfire.cache.query.CqQuery;
|
||||
public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
|
||||
|
||||
// Out-of-the-box value for the default listener handler method "handleEvent".
|
||||
public static final String ORIGINAL_DEFAULT_LISTENER_METHOD = "handleEvent";
|
||||
public static final String DEFAULT_LISTENER_METHOD_NAME = "handleEvent";
|
||||
|
||||
protected final Log logger = LogFactory.getLog(getClass());
|
||||
|
||||
@@ -82,7 +82,7 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
|
||||
|
||||
private Object delegate;
|
||||
|
||||
private String defaultListenerMethod = ORIGINAL_DEFAULT_LISTENER_METHOD;
|
||||
private String defaultListenerMethod = DEFAULT_LISTENER_METHOD_NAME;
|
||||
|
||||
/**
|
||||
* Create a new {@link ContinuousQueryListenerAdapter} with default settings.
|
||||
@@ -110,7 +110,7 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
|
||||
* @param delegate delegate object
|
||||
*/
|
||||
public void setDelegate(Object delegate) {
|
||||
Assert.notNull(delegate, "The delegate must not be null.");
|
||||
Assert.notNull(delegate, "'delegate' must not be null");
|
||||
this.delegate = delegate;
|
||||
this.invoker = null;
|
||||
}
|
||||
@@ -126,7 +126,7 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
|
||||
|
||||
/**
|
||||
* Specify the name of the default listener method to delegate to in the case where no specific listener method
|
||||
* has been determined. Out-of-the-box value is {@link #ORIGINAL_DEFAULT_LISTENER_METHOD "handleEvent}.
|
||||
* has been determined. Out-of-the-box value is {@link #DEFAULT_LISTENER_METHOD_NAME "handleEvent}.
|
||||
*
|
||||
* @param defaultListenerMethod the name of the default listener method to invoke.
|
||||
* @see #getListenerMethodName
|
||||
@@ -157,30 +157,28 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
|
||||
public void onEvent(CqEvent event) {
|
||||
try {
|
||||
// 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) {
|
||||
((ContinuousQueryListener) delegate).onEvent(event);
|
||||
return;
|
||||
// If so, this adapter will simply act as a pass-through.
|
||||
if (delegate != this && delegate instanceof ContinuousQueryListener) {
|
||||
((ContinuousQueryListener) delegate).onEvent(event);
|
||||
}
|
||||
// Else... find the listener handler method reflectively.
|
||||
else {
|
||||
String methodName = getListenerMethodName(event);
|
||||
|
||||
if (methodName == null) {
|
||||
throw new InvalidDataAccessApiUsageException("No default listener method specified."
|
||||
+ " Either specify a non-null value for the 'defaultListenerMethod' property"
|
||||
+ " or override the 'getListenerMethodName' method.");
|
||||
}
|
||||
|
||||
invoker = (invoker != null ? invoker : new MethodInvoker(delegate, methodName));
|
||||
|
||||
invokeListenerMethod(event, methodName);
|
||||
}
|
||||
|
||||
// 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);
|
||||
}
|
||||
|
||||
invokeListenerMethod(event, methodName);
|
||||
} catch (Throwable th) {
|
||||
handleListenerException(th);
|
||||
}
|
||||
catch (Throwable cause) {
|
||||
handleListenerException(cause);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -201,10 +199,10 @@ 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
|
||||
* @param cause the exception to handle
|
||||
*/
|
||||
protected void handleListenerException(Throwable ex) {
|
||||
logger.error("Listener execution failed...", ex);
|
||||
protected void handleListenerException(Throwable cause) {
|
||||
logger.error("Listener execution failed...", cause);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -223,117 +221,123 @@ public class ContinuousQueryListenerAdapter implements ContinuousQueryListener {
|
||||
}
|
||||
else {
|
||||
throw new GemfireListenerExecutionFailedException(
|
||||
String.format("Listener method '%1$s' threw exception...", methodName), e.getTargetException());
|
||||
String.format("Listener method [%1$s] threw Exception...", methodName), e.getTargetException());
|
||||
}
|
||||
}
|
||||
catch (Throwable e) {
|
||||
throw new GemfireListenerExecutionFailedException(
|
||||
String.format("Failed to invoke target listener method '%1$s'", methodName), e);
|
||||
String.format("Failed to invoke the 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();
|
||||
|
||||
this.delegate = delegate;
|
||||
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;
|
||||
return isValidEventMethodSignature(method, methodName);
|
||||
}
|
||||
});
|
||||
|
||||
Assert.isTrue(!methods.isEmpty(), "Cannot find a suitable method named [" + c.getName() + "#" + methodName
|
||||
+ "] - is the method public and has the proper arguments?");
|
||||
Assert.isTrue(!methods.isEmpty(), String.format(
|
||||
"Cannot find a suitable method named [%1$s#%2$s] - is the method public and does it have the proper arguments?",
|
||||
c.getName(), methodName));
|
||||
}
|
||||
|
||||
@SuppressWarnings("all")
|
||||
boolean isValidEventMethodSignature(Method method, String methodName) {
|
||||
if (Modifier.isPublic(method.getModifiers()) && methodName.equals(method.getName())) {
|
||||
Class<?>[] parameterTypes = method.getParameterTypes();
|
||||
|
||||
int objects = 0;
|
||||
int operations = 0;
|
||||
|
||||
if (parameterTypes.length > 0) {
|
||||
for (Class<?> parameterType : parameterTypes) {
|
||||
if (Object.class.equals(parameterType)) {
|
||||
if (++objects > 2) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
else if (Operation.class.equals(parameterType)) {
|
||||
if (++operations > 2) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
else if (byte[].class.equals(parameterType)) {
|
||||
}
|
||||
else if (CqEvent.class.equals(parameterType)) {
|
||||
}
|
||||
else if (CqQuery.class.equals(parameterType)) {
|
||||
}
|
||||
else if (Throwable.class.equals(parameterType)) {
|
||||
}
|
||||
else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
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);
|
||||
for (Method method : methods) {
|
||||
method.invoke(delegate, getMethodArguments(method, event));
|
||||
}
|
||||
}
|
||||
|
||||
Object[] getMethodArguments(Method method, CqEvent event) {
|
||||
Class<?>[] parameterTypes = method.getParameterTypes();
|
||||
Object[] args = new Object[parameterTypes.length];
|
||||
|
||||
boolean query = false;
|
||||
boolean value = false;
|
||||
|
||||
for (int index = 0; index < parameterTypes.length; index++) {
|
||||
Class<?> parameterType = parameterTypes[index];
|
||||
|
||||
if (Object.class.equals(parameterType)) {
|
||||
args[index] = (value ? event.getNewValue() : event.getKey());
|
||||
value = true;
|
||||
}
|
||||
else if (Operation.class.equals(parameterType)) {
|
||||
args[index] = (query ? event.getQueryOperation() : event.getBaseOperation());
|
||||
query = true;
|
||||
}
|
||||
else if (byte[].class.equals(parameterType)) {
|
||||
args[index] = event.getDeltaValue();
|
||||
}
|
||||
else if (CqEvent.class.equals(parameterType)) {
|
||||
args[index] = event;
|
||||
}
|
||||
else if (CqQuery.class.equals(parameterType)) {
|
||||
args[index] = event.getCq();
|
||||
}
|
||||
else if (Throwable.class.equals(parameterType)) {
|
||||
args[index] = event.getThrowable();
|
||||
}
|
||||
}
|
||||
|
||||
return args;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user