diff --git a/src/main/java/org/springframework/data/gemfire/listener/GemfireListenerExecutionFailedException.java b/src/main/java/org/springframework/data/gemfire/listener/GemfireListenerExecutionFailedException.java new file mode 100644 index 00000000..3cbab3f3 --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/listener/GemfireListenerExecutionFailedException.java @@ -0,0 +1,48 @@ +/* + * 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.listener; + +import org.springframework.dao.InvalidDataAccessApiUsageException; +import org.springframework.data.gemfire.listener.adapter.QueryListenerAdapter; + +/** + * Exception thrown when the execution of a listener method failed. + * + * @author Costin Leau + * @see QueryListenerAdapter + */ +public class GemfireListenerExecutionFailedException extends InvalidDataAccessApiUsageException { + + /** + * Constructs a new GemfireListenerExecutionFailedException instance. + * + * @param msg + * @param cause + */ + public GemfireListenerExecutionFailedException(String msg, Throwable cause) { + super(msg, cause); + } + + /** + * Constructs a new GemfireListenerExecutionFailedException instance. + * + * @param msg + */ + public GemfireListenerExecutionFailedException(String msg) { + super(msg); + } +} diff --git a/src/main/java/org/springframework/data/gemfire/listener/QueryListener.java b/src/main/java/org/springframework/data/gemfire/listener/QueryListener.java new file mode 100644 index 00000000..fdd36d1a --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/listener/QueryListener.java @@ -0,0 +1,29 @@ +/* + * 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.listener; + +import com.gemstone.gemfire.cache.query.CqEvent; + +/** + * Continuous query listener. + * + * @author Costin Leau + */ +public interface QueryListener { + + void onEvent(CqEvent event); +} diff --git a/src/main/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapter.java b/src/main/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapter.java new file mode 100644 index 00000000..1f871ec9 --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapter.java @@ -0,0 +1,341 @@ +/* + * 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.listener.adapter; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.lang.reflect.Modifier; +import java.util.ArrayList; +import java.util.List; + +import org.apache.commons.logging.Log; +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.util.Assert; +import org.springframework.util.ReflectionUtils; +import org.springframework.util.ReflectionUtils.MethodCallback; +import org.springframework.util.ReflectionUtils.MethodFilter; + +import com.gemstone.gemfire.cache.Operation; +import com.gemstone.gemfire.cache.query.CqEvent; +import com.gemstone.gemfire.cache.query.CqQuery; + +/** + * Event listener adapter that delegates the handling of messages to target + * listener methods via reflection, with flexible event type conversion. + * 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 + * Spring Framework. + * + *

By default, the content of incoming GemFire events gets extracted before + * being passed into the target listener method, to let the target method + * operate on event content types such as Object or Operation instead of + * the raw {@link CqEvent}. + * + *

Find below some examples of method signatures compliant with this + * adapter class. This first example handles all CqEvent types + * and gets passed the contents of each event type as an + * argument. + * + *

public interface PojoListener {
+ *    void handleMessage(CqEvent event);
+ *    void handleMessage(Operation op);
+ *    void handleMessage(Object key);
+ *    void handleMessage(Object key, Object newValue);
+ *    void handleMessage(Throwable th);
+ *    void handleMessage(CqEvent event, Operation op, byte[] deltaValue);
+ * }
+ * + * @author Juergen Hoeller + * @author Costin Leau + * @see org.springframework.jms.listener.adapter.MessageListenerAdapter + */ +public class QueryListenerAdapter implements QueryListener { + + 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 { + 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) { + boolean typeFound = false; + + if (Object.class.equals(paramType)) { + objects++; + if (objects > 2) { + return false; + } + typeFound = true; + } + else if (Operation.class.equals(paramType)) { + operations++; + if (operations > 2) { + return false; + } + typeFound = true; + } + else if (CqEvent.class.equals(paramType)) { + typeFound = true; + } + else if (Throwable.class.equals(paramType)) { + typeFound = true; + } + else if (byte[].class.equals(paramType)) { + typeFound = true; + } + else if (CqQuery.class.equals(paramType)) { + typeFound = true; + } + else if (!typeFound) { + return false; + } + } + } + } + return false; + } + }); + + Assert.isTrue(methods.isEmpty(), "Cannot find a suitable methods named [" + 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: "handleMessage". + */ + public static final String ORIGINAL_DEFAULT_LISTENER_METHOD = "handleMessage"; + + + /** Logger available to subclasses */ + protected final Log logger = LogFactory.getLog(getClass()); + + private Object delegate; + + private String defaultListenerMethod = ORIGINAL_DEFAULT_LISTENER_METHOD; + + private MethodInvoker invoker; + + /** + * Create a new {@link QueryListenerAdapter} with default settings. + */ + public QueryListenerAdapter() { + setDelegate(this); + } + + /** + * Create a new {@link QueryListenerAdapter} for the given delegate. + * + * @param delegate the delegate object + */ + public QueryListenerAdapter(Object delegate) { + setDelegate(delegate); + } + + + /** + * Set a target object to delegate events listening to. + * Specified listener methods have to be present on this target object. + *

If no explicit delegate object has been specified, listener + * methods are expected to present on this adapter instance, that is, + * on a custom subclass of this adapter, defining listener methods. + * + * @param delegate delegate object + */ + public void setDelegate(Object delegate) { + Assert.notNull(delegate, "Delegate must not be null"); + this.delegate = delegate; + this.invoker = null; + } + + /** + * Returns the target object to delegate event listening to. + * + * @return event listening delegation + */ + public Object getDelegate() { + return this.delegate; + } + + /** + * Specify the name of the default listener method to delegate to, + * for the case where no specific listener method has been determined. + * Out-of-the-box value is {@link #ORIGINAL_DEFAULT_LISTENER_METHOD "handleMessage"}. + * @see #getListenerMethodName + */ + public void setDefaultListenerMethod(String defaultListenerMethod) { + this.defaultListenerMethod = defaultListenerMethod; + this.invoker = null; + } + + /** + * Return the name of the default listener method to delegate to. + */ + protected String getDefaultListenerMethod() { + return this.defaultListenerMethod; + } + + /** + * Standard {@link QueryListener} entry point. + *

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. + * + * @param event the incoming GemFire event + * @see #handleListenerException + */ + @SuppressWarnings("unchecked") + public void onEvent(CqEvent event) { + try { + + // Check whether the delegate is a QueryListener 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); + } + } + + // Regular case: find a handler method reflectively. + String methodName = getListenerMethodName(event); + 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) { + handleListenerException(th); + } + } + + /** + * 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. + *

The default implementation simply returns the configured + * default listener method, if any. + * @param event the GemFire event + * @return the name of the listener method (never null) + * @see #setDefaultListenerMethod + */ + protected String getListenerMethodName(CqEvent event) { + return getDefaultListenerMethod(); + } + + /** + * Invoke the specified listener method. + * @param event the event arguments to be passed in + * @return the result returned from the listener method + * @see #getListenerMethodName + */ + 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; + } + else { + throw new GemfireListenerExecutionFailedException("Listener method '" + methodName + + "' threw exception", targetEx); + } + } catch (Throwable ex) { + throw new GemfireListenerExecutionFailedException("Failed to invoke target method '" + methodName, ex); + } + } +} \ No newline at end of file