Refine destination semantics for msg-handling methods

After this change, annotated message handling methods configured to use
a destination prefix (e.g. "/app") no longer have to include the prefix
in their mapping. For example if a client sends a message to "/app/foo"
the annotated methods should be mapped with @MessageMapping("/foo").
This commit is contained in:
Rossen Stoyanchev
2013-09-03 11:04:00 -04:00
parent e1a46bb57a
commit 0ac6998e60
20 changed files with 325 additions and 126 deletions

View File

@@ -25,6 +25,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -149,7 +150,10 @@ public final class MessageHeaders implements Map<String, Object>, Serializable {
@Override
public String toString() {
return this.headers.toString();
Map<String, Object> map = new LinkedHashMap<String, Object>(this.headers);
map.put(ID, map.remove(ID)); // remove and add again at the end
map.put(TIMESTAMP, map.remove(TIMESTAMP));
return map.toString();
}
/*

View File

@@ -19,6 +19,8 @@ package org.springframework.messaging.handler.method;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.core.MethodParameter;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
@@ -30,6 +32,8 @@ import org.springframework.util.Assert;
*/
public class HandlerMethodReturnValueHandlerComposite implements HandlerMethodReturnValueHandler {
private static Log logger = LogFactory.getLog(HandlerMethodReturnValueHandlerComposite.class);
private final List<HandlerMethodReturnValueHandler> returnValueHandlers = new ArrayList<HandlerMethodReturnValueHandler>();
@@ -61,6 +65,9 @@ public class HandlerMethodReturnValueHandlerComposite implements HandlerMethodRe
private HandlerMethodReturnValueHandler getReturnValueHandler(MethodParameter returnType) {
for (HandlerMethodReturnValueHandler handler : this.returnValueHandlers) {
if (handler.supportsReturnType(returnType)) {
if (logger.isTraceEnabled()) {
logger.trace("Processing return value with " + handler);
}
return handler;
}
}
@@ -72,7 +79,7 @@ public class HandlerMethodReturnValueHandlerComposite implements HandlerMethodRe
throws Exception {
HandlerMethodReturnValueHandler handler = getReturnValueHandler(returnType);
Assert.notNull(handler, "Unknown return value type [" + returnType.getParameterType().getName() + "]");
Assert.notNull(handler, "No handler for return value type [" + returnType.getParameterType().getName() + "]");
handler.handleReturnValue(returnValue, returnType, message);
}

View File

@@ -104,16 +104,17 @@ public class InvocableHandlerMethod extends HandlerMethod {
Object[] args = getMethodArgumentValues(message, providedArgs);
if (logger.isTraceEnabled()) {
StringBuilder builder = new StringBuilder("Invoking [");
builder.append(this.getMethod().getName()).append("] method with arguments ");
builder.append(Arrays.asList(args));
logger.trace(builder.toString());
StringBuilder sb = new StringBuilder("Invoking [");
sb.append(this.getBeanType().getSimpleName()).append(".");
sb.append(this.getMethod().getName()).append("] method with arguments ");
sb.append(Arrays.asList(args));
logger.trace(sb.toString());
}
Object returnValue = invoke(args);
if (logger.isTraceEnabled()) {
logger.trace("Method [" + this.getMethod().getName() + "] returned [" + returnValue + "]");
logger.trace("Method returned [" + returnValue + "]");
}
return returnValue;

View File

@@ -16,7 +16,10 @@
package org.springframework.messaging.simp.annotation.support;
import java.lang.annotation.Annotation;
import org.springframework.core.MethodParameter;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.MessagePostProcessor;
@@ -27,6 +30,7 @@ import org.springframework.messaging.simp.SimpMessageSendingOperations;
import org.springframework.messaging.simp.annotation.ReplyToUser;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
/**
@@ -46,17 +50,23 @@ public class ReplyToMethodReturnValueHandler implements HandlerMethodReturnValue
private final SimpMessageSendingOperations messagingTemplate;
private final boolean annotationRequired;
public ReplyToMethodReturnValueHandler(SimpMessageSendingOperations messagingTemplate) {
public ReplyToMethodReturnValueHandler(SimpMessageSendingOperations messagingTemplate, boolean annotationRequired) {
Assert.notNull(messagingTemplate, "messagingTemplate is required");
this.messagingTemplate = messagingTemplate;
this.annotationRequired = annotationRequired;
}
@Override
public boolean supportsReturnType(MethodParameter returnType) {
return ((returnType.getMethodAnnotation(ReplyTo.class) != null)
|| (returnType.getMethodAnnotation(ReplyToUser.class) != null));
if ((returnType.getMethodAnnotation(ReplyTo.class) != null) ||
(returnType.getMethodAnnotation(ReplyToUser.class) != null)) {
return true;
}
return (!this.annotationRequired);
}
@Override
@@ -72,23 +82,32 @@ public class ReplyToMethodReturnValueHandler implements HandlerMethodReturnValue
String sessionId = inputHeaders.getSessionId();
MessagePostProcessor postProcessor = new SessionHeaderPostProcessor(sessionId);
ReplyTo replyTo = returnType.getMethodAnnotation(ReplyTo.class);
if (replyTo != null) {
for (String destination : replyTo.value()) {
this.messagingTemplate.convertAndSend(destination, returnValue, postProcessor);
}
}
ReplyToUser replyToUser = returnType.getMethodAnnotation(ReplyToUser.class);
if (replyToUser != null) {
if (inputHeaders.getUser() == null) {
throw new MissingSessionUserException(inputMessage);
}
String user = inputHeaders.getUser().getName();
for (String destination : replyToUser.value()) {
for (String destination : getDestinations(replyToUser, inputHeaders.getDestination())) {
this.messagingTemplate.convertAndSendToUser(user, destination, returnValue, postProcessor);
}
return;
}
ReplyTo replyTo = returnType.getMethodAnnotation(ReplyTo.class);
if (replyTo != null) {
for (String destination : getDestinations(replyTo, inputHeaders.getDestination())) {
this.messagingTemplate.convertAndSend(destination, returnValue, postProcessor);
}
return;
}
this.messagingTemplate.convertAndSend(inputHeaders.getDestination(), returnValue, postProcessor);
}
private String[] getDestinations(Annotation annot, String inputDestination) {
String[] destinations = (String[]) AnnotationUtils.getValue(annot);
return ObjectUtils.isEmpty(destinations) ? new String[] { inputDestination } : destinations;
}
@@ -107,4 +126,10 @@ public class ReplyToMethodReturnValueHandler implements HandlerMethodReturnValue
return MessageBuilder.withPayloadAndHeaders(message.getPayload(), headers).build();
}
}
@Override
public String toString() {
return "ReplyToMethodReturnValueHandler [annotationRequired=" + annotationRequired + "]";
}
}

View File

@@ -47,16 +47,37 @@ public class MessageBrokerConfigurer {
this.webSocketResponseChannel = webSocketResponseChannel;
}
/**
* Enable a simple message broker and configure one or more prefixes to filter
* destinations targeting the broker (e.g. destinations prefixed with "/topic").
*/
public SimpleBrokerRegistration enableSimpleBroker(String... destinationPrefixes) {
this.simpleBroker = new SimpleBrokerRegistration(this.webSocketResponseChannel, destinationPrefixes);
return this.simpleBroker;
}
/**
* Enable a STOMP broker relay and configure the destination prefixes supported by the
* message broker. Check the STOMP documentation of the message broker for supported
* destinations.
*/
public StompBrokerRelayRegistration enableStompBrokerRelay(String... destinationPrefixes) {
this.stompRelay = new StompBrokerRelayRegistration(this.webSocketResponseChannel, destinationPrefixes);
return this.stompRelay;
}
/**
* Configure one or more prefixes to filter destinations targeting annotated
* application methods. For example destinations prefixed with "/app" may be processed
* by annotated application methods while other destinations may target the message
* broker (e.g. "/topic", "/queue").
* <p>
* When messages are processed, the matching prefix is removed from the destination in
* order to form the lookup path. This means annotations should not contain the
* destination prefix.
* <p>
* Prefixes that do not have a trailing slash will have one automatically appended.
*/
public MessageBrokerConfigurer setAnnotationMethodDestinationPrefixes(String... destinationPrefixes) {
this.annotationMethodDestinationPrefixes = destinationPrefixes;
return this;

View File

@@ -141,7 +141,7 @@ public abstract class AbstractBrokerMessageHandler
}
if (logger.isTraceEnabled()) {
logger.trace("Processing message: " + message);
logger.trace("Message " + message);
}
handleMessageInternal(message);

View File

@@ -56,7 +56,8 @@ public abstract class AbstractSubscriptionRegistry implements SubscriptionRegist
return;
}
if (logger.isDebugEnabled()) {
logger.debug("Subscribe request: " + message);
logger.debug("Adding subscription id=" + headers.getSubscriptionId()
+ ", destination=" + headers.getDestination());
}
addSubscriptionInternal(sessionId, subscriptionId, destination, message);
}

View File

@@ -60,6 +60,7 @@ import org.springframework.messaging.simp.annotation.UnsubscribeEvent;
import org.springframework.messaging.simp.annotation.support.PrincipalMethodArgumentResolver;
import org.springframework.messaging.simp.annotation.support.ReplyToMethodReturnValueHandler;
import org.springframework.messaging.simp.annotation.support.SubscriptionMethodReturnValueHandler;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.messaging.support.converter.MessageConverter;
import org.springframework.stereotype.Controller;
import org.springframework.util.Assert;
@@ -80,7 +81,7 @@ public class AnnotationMethodMessageHandler implements MessageHandler, Applicati
private final SimpMessageSendingOperations webSocketResponseTemplate;
private Collection<String> destinationPrefixes;
private Collection<String> destinationPrefixes = new ArrayList<String>();
private MessageConverter<?> messageConverter;
@@ -117,9 +118,29 @@ public class AnnotationMethodMessageHandler implements MessageHandler, Applicati
this.webSocketResponseTemplate = new SimpMessagingTemplate(webSocketResponseChannel);
}
/**
* Configure one or more prefixes to filter destinations targeting annotated
* application methods. For example destinations prefixed with "/app" may be processed
* by annotated application methods while other destinations may target the message
* broker (e.g. "/topic", "/queue").
* <p>
* When messages are processed, the matching prefix is removed from the destination in
* order to form the lookup path. This means annotations should not contain the
* destination prefix.
* <p>
* Prefixes that do not have a trailing slash will have one automatically appended.
*/
public void setDestinationPrefixes(Collection<String> destinationPrefixes) {
this.destinationPrefixes = destinationPrefixes;
this.destinationPrefixes.clear();
if (destinationPrefixes != null) {
for (String prefix : destinationPrefixes) {
prefix = prefix.trim();
if (!prefix.endsWith("/")) {
prefix += "/";
}
this.destinationPrefixes.add(prefix);
}
}
}
public Collection<String> getDestinationPrefixes() {
@@ -180,14 +201,17 @@ public class AnnotationMethodMessageHandler implements MessageHandler, Applicati
this.argumentResolvers.addResolver(new MessageBodyMethodArgumentResolver(this.messageConverter));
// Annotation-based return value types
this.returnValueHandlers.addHandler(new ReplyToMethodReturnValueHandler(this.brokerTemplate));
this.returnValueHandlers.addHandler(new ReplyToMethodReturnValueHandler(this.brokerTemplate, true));
this.returnValueHandlers.addHandler(new SubscriptionMethodReturnValueHandler(this.webSocketResponseTemplate));
// custom return value types
this.returnValueHandlers.addHandlers(this.customReturnValueHandlers);
// catch-all
this.returnValueHandlers.addHandler(new ReplyToMethodReturnValueHandler(this.brokerTemplate, false));
}
protected void initHandlerMethods() {
protected final void initHandlerMethods() {
String[] beanNames = this.applicationContext.getBeanNamesForType(Object.class);
for (String beanName : beanNames) {
if (isHandler(this.applicationContext.getType(beanName))){
@@ -200,26 +224,20 @@ public class AnnotationMethodMessageHandler implements MessageHandler, Applicati
return (AnnotationUtils.findAnnotation(beanType, Controller.class) != null);
}
protected void detectHandlerMethods(Object handler) {
protected final void detectHandlerMethods(Object handler) {
Class<?> handlerType = (handler instanceof String) ?
this.applicationContext.getType((String) handler) : handler.getClass();
final Class<?> userType = ClassUtils.getUserClass(handlerType);
handlerType = ClassUtils.getUserClass(handlerType);
initHandlerMethods(handler, userType, MessageMapping.class,
new MessageMappingInfoCreator(), this.messageMethods);
initHandlerMethods(handler, userType, SubscribeEvent.class,
new SubscribeMappingInfoCreator(), this.subscribeMethods);
initHandlerMethods(handler, userType, UnsubscribeEvent.class,
new UnsubscribeMappingInfoCreator(), this.unsubscribeMethods);
initHandlerMethods(handler, handlerType, MessageMapping.class, this.messageMethods);
initHandlerMethods(handler, handlerType, SubscribeEvent.class, this.subscribeMethods);
initHandlerMethods(handler, handlerType, UnsubscribeEvent.class, this.unsubscribeMethods);
}
private <A extends Annotation> void initHandlerMethods(Object handler, Class<?> handlerType,
final Class<A> annotationType, MappingInfoCreator<A> mappingInfoCreator,
Map<MappingInfo, HandlerMethod> handlerMethods) {
final Class<A> annotationType, Map<MappingInfo, HandlerMethod> handlerMethods) {
Set<Method> methods = HandlerMethodSelector.selectMethods(handlerType, new MethodFilter() {
@Override
@@ -230,12 +248,25 @@ public class AnnotationMethodMessageHandler implements MessageHandler, Applicati
for (Method method : methods) {
A annotation = AnnotationUtils.findAnnotation(method, annotationType);
HandlerMethod hm = createHandlerMethod(handler, method);
handlerMethods.put(mappingInfoCreator.create(annotation), hm);
String[] destinations = (String[]) AnnotationUtils.getValue(annotation);
MappingInfo mapping = new MappingInfo(destinations);
HandlerMethod newHandlerMethod = createHandlerMethod(handler, method);
HandlerMethod oldHandlerMethod = handlerMethods.get(mapping);
if (oldHandlerMethod != null && !oldHandlerMethod.equals(newHandlerMethod)) {
throw new IllegalStateException("Ambiguous mapping found. Cannot map '" + newHandlerMethod.getBean()
+ "' bean method \n" + newHandlerMethod + "\nto " + mapping + ": There is already '"
+ oldHandlerMethod.getBean() + "' bean method\n" + oldHandlerMethod + " mapped.");
}
handlerMethods.put(mapping, newHandlerMethod);
if (logger.isInfoEnabled()) {
logger.info("Mapped \"@" + annotationType.getSimpleName()
+ " " + mapping + "\" onto " + newHandlerMethod);
}
}
}
protected HandlerMethod createHandlerMethod(Object handler, Method method) {
private HandlerMethod createHandlerMethod(Object handler, Method method) {
HandlerMethod handlerMethod;
if (handler instanceof String) {
String beanName = (String) handler;
@@ -264,22 +295,27 @@ public class AnnotationMethodMessageHandler implements MessageHandler, Applicati
}
}
private void handleMessageInternal(final Message<?> message, Map<MappingInfo, HandlerMethod> handlerMethods) {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
String destination = headers.getDestination();
if (!checkDestinationPrefix(destination)) {
return;
}
HandlerMethod match = getHandlerMethod(destination, handlerMethods);
if (match == null) {
return;
}
private void handleMessageInternal(Message<?> message, Map<MappingInfo, HandlerMethod> handlerMethods) {
if (logger.isTraceEnabled()) {
logger.trace("Processing message: " + message);
logger.trace("Message " + message);
}
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
String lookupPath = getLookupPath(headers.getDestination());
if (lookupPath == null) {
if (logger.isTraceEnabled()) {
logger.trace("Ignoring message with destination " + headers.getDestination());
}
return;
}
HandlerMethod match = getHandlerMethod(lookupPath, handlerMethods);
if (match == null) {
if (logger.isTraceEnabled()) {
logger.trace("No matching method, lookup path " + lookupPath);
}
return;
}
HandlerMethod handlerMethod = match.createWithResolvedBean();
@@ -288,6 +324,9 @@ public class AnnotationMethodMessageHandler implements MessageHandler, Applicati
invocableHandlerMethod.setMessageMethodArgumentResolvers(this.argumentResolvers);
try {
headers.setDestination(lookupPath);
message = MessageBuilder.withPayloadAndHeaders(message.getPayload(), headers).build();
Object returnValue = invocableHandlerMethod.invoke(message);
MethodParameter returnType = handlerMethod.getReturnType();
@@ -305,16 +344,18 @@ public class AnnotationMethodMessageHandler implements MessageHandler, Applicati
}
}
private boolean checkDestinationPrefix(String destination) {
if ((destination == null) || CollectionUtils.isEmpty(this.destinationPrefixes)) {
return true;
}
for (String prefix : this.destinationPrefixes) {
if (destination.startsWith(prefix)) {
return true;
private String getLookupPath(String destination) {
if (destination != null) {
if (CollectionUtils.isEmpty(this.destinationPrefixes)) {
return destination;
}
for (String prefix : this.destinationPrefixes) {
if (destination.startsWith(prefix)) {
return destination.substring(prefix.length() - 1);
}
}
}
return false;
return null;
}
private void invokeExceptionHandler(Message<?> message, HandlerMethod handlerMethod, Exception ex) {
@@ -365,49 +406,38 @@ public class AnnotationMethodMessageHandler implements MessageHandler, Applicati
private static class MappingInfo {
private final List<String> destinations;
private final String[] destinations;
public MappingInfo(List<String> destinations) {
public MappingInfo(String[] destinations) {
Assert.notNull(destinations, "No destinations");
this.destinations = destinations;
}
public List<String> getDestinations() {
public String[] getDestinations() {
return this.destinations;
}
@Override
public int hashCode() {
return Arrays.hashCode(this.destinations);
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o != null && getClass().equals(o.getClass())) {
MappingInfo other = (MappingInfo) o;
return Arrays.equals(destinations, other.getDestinations());
}
return false;
}
@Override
public String toString() {
return "MappingInfo [destinations=" + this.destinations + "]";
}
}
private interface MappingInfoCreator<A extends Annotation> {
MappingInfo create(A annotation);
}
private static class MessageMappingInfoCreator implements MappingInfoCreator<MessageMapping> {
@Override
public MappingInfo create(MessageMapping annotation) {
return new MappingInfo(Arrays.asList(annotation.value()));
}
}
private static class SubscribeMappingInfoCreator implements MappingInfoCreator<SubscribeEvent> {
@Override
public MappingInfo create(SubscribeEvent annotation) {
return new MappingInfo(Arrays.asList(annotation.value()));
}
}
private static class UnsubscribeMappingInfoCreator implements MappingInfoCreator<UnsubscribeEvent> {
@Override
public MappingInfo create(UnsubscribeEvent annotation) {
return new MappingInfo(Arrays.asList(annotation.value()));
return "[destinations=" + Arrays.toString(this.destinations) + "]";
}
}

View File

@@ -80,34 +80,27 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
String destination = headers.getDestination();
if (!checkDestinationPrefix(destination)) {
if (logger.isTraceEnabled()) {
logger.trace("Ingoring message with destination " + destination);
}
return;
}
if (SimpMessageType.SUBSCRIBE.equals(messageType)) {
preProcessMessage(message);
this.subscriptionRegistry.registerSubscription(message);
}
else if (SimpMessageType.UNSUBSCRIBE.equals(messageType)) {
preProcessMessage(message);
this.subscriptionRegistry.unregisterSubscription(message);
}
else if (SimpMessageType.MESSAGE.equals(messageType)) {
preProcessMessage(message);
sendMessageToSubscribers(headers.getDestination(), message);
}
else if (SimpMessageType.DISCONNECT.equals(messageType)) {
preProcessMessage(message);
String sessionId = SimpMessageHeaderAccessor.wrap(message).getSessionId();
this.subscriptionRegistry.unregisterAllSubscriptions(sessionId);
}
}
private void preProcessMessage(Message<?> message) {
if (logger.isTraceEnabled()) {
logger.trace("Processing " + message);
}
}
protected void sendMessageToSubscribers(String destination, Message<?> message) {
MultiValueMap<String,String> subscriptions = this.subscriptionRegistry.findSubscriptions(message);
for (String sessionId : subscriptions.keySet()) {

View File

@@ -103,7 +103,7 @@ public class StompProtocolHandler implements SubProtocolHandler {
// http://stomp.github.io/stomp-specification-1.2.html#Size_Limits
if (logger.isTraceEnabled()) {
logger.trace("Processing STOMP message: " + message);
logger.trace("Message " + message);
}
try {

View File

@@ -82,7 +82,10 @@ public class GenericMessage<T> implements Message<T>, Serializable {
}
public String toString() {
return "[Payload=" + this.payload + "][Headers=" + this.headers + "]";
StringBuilder sb = new StringBuilder("[Headers=" + this.headers + "]");
sb.append("[Payload ").append(this.payload.getClass().getSimpleName());
sb.append(" content=").append(this.payload).append("]");
return sb.toString();
}
public int hashCode() {

View File

@@ -103,7 +103,7 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName
Assert.notNull(message, "Message must not be null");
if (logger.isTraceEnabled()) {
logger.trace("[" + this.beanName + "] sending message " + message);
logger.trace("[" + this.beanName + "] send message " + message);
}
message = this.interceptorChain.preSend(message, this);

View File

@@ -18,6 +18,7 @@ package org.springframework.messaging.support.channel;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.commons.logging.Log;
@@ -57,21 +58,30 @@ class ChannelInterceptorChain {
public Message<?> preSend(Message<?> message, MessageChannel channel) {
UUID originalId = message.getHeaders().getId();
if (logger.isTraceEnabled()) {
logger.trace("preSend on channel '" + channel + "', message: " + message);
logger.trace("preSend message id " + originalId);
}
for (ChannelInterceptor interceptor : this.interceptors) {
message = interceptor.preSend(message, channel);
if (message == null) {
if (logger.isTraceEnabled()) {
logger.trace("preSend returned null (precluding the send)");
}
return null;
}
}
if (logger.isTraceEnabled()) {
if (!message.getHeaders().getId().equals(originalId)) {
logger.trace("preSend returned modified message " + message);
}
}
return message;
}
public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
if (logger.isTraceEnabled()) {
logger.trace("postSend (sent=" + sent + ") on channel '" + channel + "', message: " + message);
logger.trace("postSend (sent=" + sent + ") message id " + message.getHeaders().getId());
}
for (ChannelInterceptor interceptor : this.interceptors) {
interceptor.postSend(message, channel, sent);

View File

@@ -64,6 +64,9 @@ public class ExecutorSubscribableChannel extends AbstractSubscribableChannel {
@Override
public boolean sendInternal(final Message<?> message, long timeout) {
logger.trace("subscribers " + this.handlers);
for (final MessageHandler handler : this.handlers) {
if (this.executor == null) {
handler.handleMessage(message);