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

DATAREDIS-77

Browse Source
This commit is contained in:
Costin Leau
2011-10-27 21:50:27 +03:00
parent 9c5d01e9a2
commit 62a5588af4
2 changed files with 229 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

296
src/main/java/org/springframework/data/redis/listener/adapter/MessageListenerAdapter.java
View File

@@ -16,6 +16,10 @@
package org.springframework.data.redis.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;
@@ -27,8 +31,10 @@ import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.util.Assert;
import org.springframework.util.MethodInvoker;
import org.springframework.util.ObjectUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.ReflectionUtils.MethodCallback;
import org.springframework.util.ReflectionUtils.MethodFilter;
/**
* Message listener adapter that delegates the handling of messages to target
@@ -36,58 +42,132 @@ import org.springframework.util.ObjectUtils;
* Allows listener methods to operate on message content types, completely
* independent from the Redis API.
*
* <p/>Modeled as much as possible after the JMS MessageListenerAdapter in
* Spring Framework.
*
* <p>By default, the content of incoming Redis messages gets extracted before
* <p/>
* Modeled as much as possible after the JMS MessageListenerAdapter in Spring
* Framework.
*
* <p>
* By default, the content of incoming Redis messages gets extracted before
* being passed into the target listener method, to let the target method
* operate on message content types such as String or byte array instead of
* the raw {@link Message}. Message type conversion is delegated to a Spring
* Data {@link RedisSerializer}. By default, the {@link JdkSerializationRedisSerializer}
* will be used. (If you do not want such automatic message conversion taking
* place, then be sure to set the {@link #setSerializer Serializer}
* to <code>null</code>.)
*
* <p>Find below some examples of method signatures compliant with this
* adapter class. This first example handles all <code>Message</code> types
* and gets passed the contents of each <code>Message</code> type as an
* argument.
*
* <pre class="code">public interface MessageContentsDelegate {
* void handleMessage(String text);
* void handleMessage(byte[] bytes);
* void handleMessage(Person obj);
* }</pre>
* operate on message content types such as String or byte array instead of the
* raw {@link Message}. Message type conversion is delegated to a Spring Data
* {@link RedisSerializer}. By default, the
* {@link JdkSerializationRedisSerializer} will be used. (If you do not want
* such automatic message conversion taking place, then be sure to set the
* {@link #setSerializer Serializer} to <code>null</code>.)
*
* <p>
* Find below some examples of method signatures compliant with this adapter
* class. This first example handles all <code>Message</code> types and gets
* passed the contents of each <code>Message</code> type as an argument.
*
* <pre class="code">
* public interface MessageContentsDelegate {
* void handleMessage(String text);
*
* void handleMessage(byte[] bytes);
*
* void handleMessage(Person obj);
* }
* </pre>
*
* <p>
* In addition, the channel or pattern to which a message is sent can be passed in
* to the method as a second argument of type String:
*
* <pre class="code">
* public interface MessageContentsDelegate {
* void handleMessage(String text, String channel);
*
* void handleMessage(byte[] bytes, String pattern);
* }
* </pre>
*
*
* For further examples and discussion please do refer to the Spring Data
* reference documentation which describes this class (and its attendant
* configuration) in detail.
*
* <b>Important:</b> Due to the nature of messages, the default serializer used by
* the adapter is {@link StringRedisSerializer}. If the messages are of a different type,
* change them accordingly through {@link #setSerializer(RedisSerializer)}.
*
* <b>Important:</b> Due to the nature of messages, the default serializer used
* by the adapter is {@link StringRedisSerializer}. If the messages are of a
* different type, change them accordingly through
* {@link #setSerializer(RedisSerializer)}.
*
* @author Juergen Hoeller
* @author Costin Leau
* @see org.springframework.jms.listener.adapter.MessageListenerAdapter
*/
public class MessageListenerAdapter implements MessageListener {
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 argument numbers
Class<?>[] parameterTypes = method.getParameterTypes();
return ((parameterTypes.length == 2 && String.class
.equals(parameterTypes[1])) || parameterTypes.length == 1);
}
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(Object[] arguments) throws InvocationTargetException, IllegalAccessException {
Object[] message = new Object[] {arguments[0]};
for (Method m : methods) {
Class<?>[] types = m.getParameterTypes();
Object[] args = (types.length == 2 ? arguments: message);
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 MethodInvoker invoker;
private String defaultListenerMethod = ORIGINAL_DEFAULT_LISTENER_METHOD;
private RedisSerializer<?> serializer;
private RedisSerializer<String> stringSerializer;
/**
* Create a new {@link MessageListenerAdapter} with default settings.
@@ -100,26 +180,29 @@ public class MessageListenerAdapter implements MessageListener {
/**
* Create a new {@link MessageListenerAdapter} for the given delegate.
*
* @param delegate the delegate object
* @param delegate
* the delegate object
*/
public MessageListenerAdapter(Object delegate) {
initDefaultStrategies();
setDelegate(delegate);
}
/**
* Set a target object to delegate message listening to.
* Specified listener methods have to be present on this target object.
* <p>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.
* Set a target object to delegate message listening to. Specified listener
* methods have to be present on this target object.
* <p>
* 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
* @param delegate
* delegate object
*/
public void setDelegate(Object delegate) {
Assert.notNull(delegate, "Delegate must not be null");
this.delegate = delegate;
this.invoker = null;
}
/**
@@ -132,13 +215,16 @@ public class MessageListenerAdapter implements MessageListener {
}
/**
* 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"}.
* 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;
}
/**
@@ -151,26 +237,41 @@ public class MessageListenerAdapter implements MessageListener {
/**
* Set the serializer that will convert incoming raw Redis messages to
* listener method arguments.
* <p>The default converter is a {@link StringRedisSerializer}.
* <p>
* The default converter is a {@link StringRedisSerializer}.
*
* @param serializer
*/
public void setSerializer(RedisSerializer<?> serializer) {
this.serializer = serializer;
}
/**
* Sets the serializer used for converting the channel/pattern to a String.
*
* <p>
* The default converter is a {@link StringRedisSerializer}.
*
* @param serializer
*/
public void setStringSerializer(RedisSerializer<String> serializer) {
this.stringSerializer = serializer;
}
/**
* Standard Redis {@link MessageListener} entry point.
* <p>Delegates the message to the target listener method, with appropriate
* <p>
* Delegates the message to the target listener method, with appropriate
* conversion of the message argument. In case of an exception, the
* {@link #handleListenerException(Throwable)} method will be invoked.
*
* @param message the incoming Redis message
* @param message
* the incoming Redis message
* @see #handleListenerException
*/
@Override
@SuppressWarnings("unchecked")
public void onMessage(Message message, byte[] pattern) {
try {
// Check whether the delegate is a MessageListener impl itself.
// In that case, the adapter will simply act as a pass-through.
if (delegate != this) {
@@ -181,15 +282,25 @@ public class MessageListenerAdapter implements MessageListener {
// Regular case: find a handler method reflectively.
Object convertedMessage = extractMessage(message);
String convertedChannel = stringSerializer.deserialize(pattern);
// Invoke the handler method with appropriate arguments.
Object[] listenerArguments = new Object[] { convertedMessage,
convertedChannel };
String methodName = getListenerMethodName(message, convertedMessage);
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.");
throw new InvalidDataAccessApiUsageException(
"No default listener method specified: "
+ "Either specify a non-null value for the 'defaultListenerMethod' property or "
+ "override the 'getListenerMethodName' method.");
}
// Invoke the handler method with appropriate arguments.
Object[] listenerArguments = buildListenerArguments(convertedMessage);
if (invoker == null) {
invoker = new MethodInvoker(delegate, methodName);
}
invokeListenerMethod(methodName, listenerArguments);
} catch (Throwable th) {
handleListenerException(th);
@@ -203,13 +314,17 @@ public class MessageListenerAdapter implements MessageListener {
* @see JdkSerializationRedisSerializer
*/
protected void initDefaultStrategies() {
setSerializer(new StringRedisSerializer());
RedisSerializer<String> serializer = new StringRedisSerializer();
setSerializer(serializer);
setStringSerializer(serializer);
}
/**
* Handle the given exception that arose during listener execution.
* The default implementation logs the exception at error level.
* @param ex the exception to handle
* 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);
@@ -217,9 +332,11 @@ public class MessageListenerAdapter implements MessageListener {
/**
* Extract the message body from the given Redis message.
* @param message the Redis <code>Message</code>
* @return the content of the message, to be passed into the
* listener method as argument
*
* @param message
* the Redis <code>Message</code>
* @return the content of the message, to be passed into the listener method
* as argument
*/
protected Object extractMessage(Message message) {
if (serializer != null) {
@@ -229,67 +346,52 @@ public class MessageListenerAdapter implements MessageListener {
}
/**
* Determine the name of the listener method that is supposed to
* handle the given message.
* <p>The default implementation simply returns the configured
* default listener method, if any.
* @param originalMessage the Redis request message
* @param extractedMessage the converted Redis request message,
* to be passed into the listener method as argument
* Determine the name of the listener method that is supposed to handle the
* given message.
* <p>
* The default implementation simply returns the configured default listener
* method, if any.
*
* @param originalMessage
* the Redis request message
* @param extractedMessage
* the converted Redis request message, to be passed into the
* listener method as argument
* @return the name of the listener method (never <code>null</code>)
* @see #setDefaultListenerMethod
*/
protected String getListenerMethodName(Message originalMessage, Object extractedMessage) {
protected String getListenerMethodName(Message originalMessage,
Object extractedMessage) {
return getDefaultListenerMethod();
}
/**
* Build an array of arguments to be passed into the target listener method.
* Allows for multiple method arguments to be built from a single message object.
* <p>The default implementation builds an array with the given message object
* as sole element. This means that the extracted message will always be passed
* into a <i>single</i> method argument, even if it is an array, with the target
* method having a corresponding single argument of the array's type declared.
* <p>This can be overridden to treat special message content such as arrays
* differently, for example passing in each element of the message array
* as distinct method argument.
* @param extractedMessage the content of the message
* @return the array of arguments to be passed into the
* listener method (each element of the array corresponding
* to a distinct method argument)
*/
protected Object[] buildListenerArguments(Object extractedMessage) {
return new Object[] { extractedMessage };
}
/**
* Invoke the specified listener method.
* @param methodName the name of the listener method
* @param arguments the message arguments to be passed in
* @return the result returned from the listener method
*
* @param methodName
* the name of the listener method
* @param arguments
* the message arguments to be passed in
* @see #getListenerMethodName
* @see #buildListenerArguments
*/
protected Object invokeListenerMethod(String methodName, Object[] arguments) {
protected void invokeListenerMethod(String methodName, Object[] arguments) {
try {
MethodInvoker methodInvoker = new MethodInvoker();
methodInvoker.setTargetObject(getDelegate());
methodInvoker.setTargetMethod(methodName);
methodInvoker.setArguments(arguments);
methodInvoker.prepare();
return methodInvoker.invoke();
invoker.invoke(arguments);
} catch (InvocationTargetException ex) {
Throwable targetEx = ex.getTargetException();
if (targetEx instanceof DataAccessException) {
throw (DataAccessException) targetEx;
}
else {
throw new RedisListenerExecutionFailedException("Listener method '" + methodName + "' threw exception",
} else {
throw new RedisListenerExecutionFailedException(
"Listener method '" + methodName + "' threw exception",
targetEx);
}
} catch (Throwable ex) {
throw new RedisListenerExecutionFailedException("Failed to invoke target method '" + methodName
+ "' with arguments " + ObjectUtils.nullSafeToString(arguments), ex);
throw new RedisListenerExecutionFailedException(
"Failed to invoke target method '" + methodName
+ "' with arguments "
+ ObjectUtils.nullSafeToString(arguments), ex);
}
}
}