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

Make use of enhanced MessageHeaderAccessor support

Browse Source 
Mutate rather than re-create headers when decoding STOMP messages
before a message is sent on a message channel.

Use MessageBuilder.createMessage to ensure the fully prepared
MessageHeaders is used directly MessageHeaderAccessor instance.

Issue: SPR-11468
This commit is contained in:
Rossen Stoyanchev
2014-04-10 23:57:45 -04:00
parent 4867546aec
commit ae942ffdb8
22 changed files with 322 additions and 236 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-messaging/src/main/java/org/springframework/messaging/core/AbstractMessageSendingTemplate.java
View File

@@ -26,6 +26,7 @@ import org.springframework.messaging.MessagingException;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.converter.SimpleMessageConverter;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.Assert;
/**
@@ -130,6 +131,13 @@ public abstract class AbstractMessageSendingTemplate<D> implements MessageSendin
MessagePostProcessor postProcessor) throws MessagingException {
headers = processHeadersToSend(headers);
MessageHeaders messageHeaders;
if (headers != null && headers instanceof MessageHeaders) {
MessageHeaderAccessor.getAccessor()
}
MessageHeaders messageHeaders = (headers != null) ? new MessageHeaders(headers) : null;
Message<?> message = this.converter.toMessage(payload, messageHeaders);

21
spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/support/HeadersMethodArgumentResolver.java
View File

@@ -26,7 +26,7 @@ import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.ClassUtils;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
/**
@@ -42,7 +42,6 @@ import org.springframework.util.ReflectionUtils;
*/
public class HeadersMethodArgumentResolver implements HandlerMethodArgumentResolver {
@Override
public boolean supportsParameter(MethodParameter parameter) {
Class<?> paramType = parameter.getParameterType();
@@ -60,15 +59,23 @@ public class HeadersMethodArgumentResolver implements HandlerMethodArgumentResol
return message.getHeaders();
}
else if (MessageHeaderAccessor.class.equals(paramType)) {
return new MessageHeaderAccessor(message);
MessageHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, MessageHeaderAccessor.class);
return (accessor != null ? accessor : new MessageHeaderAccessor(message));
}
else if (MessageHeaderAccessor.class.isAssignableFrom(paramType)) {
Method factoryMethod = ClassUtils.getMethod(paramType, "wrap", Message.class);
return ReflectionUtils.invokeMethod(factoryMethod, null, message);
MessageHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, MessageHeaderAccessor.class);
if (accessor != null && paramType.isAssignableFrom(accessor.getClass())) {
return accessor;
}
else {
Method method = ReflectionUtils.findMethod(paramType, "wrap", Message.class);
Assert.notNull(method, "Cannot create accessor of type " + paramType + " for message " + message);
return ReflectionUtils.invokeMethod(method, null, message);
}
}
else {
throw new IllegalStateException("Unexpected method parameter type "
+ paramType + "in method " + parameter.getMethod() + ". "
throw new IllegalStateException(
"Unexpected method parameter type " + paramType + "in method " + parameter.getMethod() + ". "
+ "@Headers method arguments must be assignable to java.util.Map.");
}
}

6
spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/AbstractMethodMessageHandler.java
View File

@@ -42,6 +42,7 @@ import org.springframework.messaging.handler.DestinationPatternsMessageCondition
import org.springframework.messaging.handler.HandlerMethod;
import org.springframework.messaging.handler.HandlerMethodSelector;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
@@ -339,8 +340,9 @@ public abstract class AbstractMethodMessageHandler<T>
logger.debug("Handling message, lookupDestination=" + lookupDestination);
}
message = MessageBuilder.fromMessage(message).setHeader(
DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER, lookupDestination).build();
MessageHeaderAccessor headerAccessor = MessageHeaderAccessor.getMutableAccessor(message);
headerAccessor.setHeader(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER, lookupDestination);
message = MessageBuilder.createMessage(message.getPayload(), headerAccessor.getMessageHeaders());
handleMessageInternal(message, lookupDestination);
}

4
spring-messaging/src/main/java/org/springframework/messaging/simp/SimpMessageTypeMessageCondition.java
View File

@@ -73,7 +73,7 @@ public class SimpMessageTypeMessageCondition extends AbstractMessageCondition<Si
@Override
public SimpMessageTypeMessageCondition getMatchingCondition(Message<?> message) {
Object actualMessageType = message.getHeaders().get(SimpMessageHeaderAccessor.MESSAGE_TYPE_HEADER);
Object actualMessageType = SimpMessageHeaderAccessor.getMessageType(message.getHeaders());
if (actualMessageType == null) {
return null;
}
@@ -83,7 +83,7 @@ public class SimpMessageTypeMessageCondition extends AbstractMessageCondition<Si
@Override
public int compareTo(SimpMessageTypeMessageCondition other, Message<?> message) {
Object actualMessageType = message.getHeaders().get(SimpMessageHeaderAccessor.MESSAGE_TYPE_HEADER);
Object actualMessageType = SimpMessageHeaderAccessor.getMessageType(message.getHeaders());
if (actualMessageType != null) {
if (actualMessageType.equals(this.getMessageType()) && actualMessageType.equals(other.getMessageType())) {
return 0;

11
spring-messaging/src/main/java/org/springframework/messaging/simp/SimpMessagingTemplate.java
View File

@@ -108,8 +108,7 @@ public class SimpMessagingTemplate extends AbstractMessageSendingTemplate<String
@Override
public void send(Message<?> message) {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
String destination = headers.getDestination();
String destination = SimpMessageHeaderAccessor.getDestination(message.getHeaders());
destination = (destination != null) ? destination : getRequiredDefaultDestination();
doSend(destination, message);
}
@@ -118,10 +117,10 @@ public class SimpMessagingTemplate extends AbstractMessageSendingTemplate<String
protected void doSend(String destination, Message<?> message) {
Assert.notNull(destination, "Destination must not be null");
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
headers.setDestination(destination);
headers.setMessageTypeIfNotSet(SimpMessageType.MESSAGE);
message = MessageBuilder.withPayload(message.getPayload()).setHeaders(headers).build();
SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.wrap(message);
headerAccessor.setDestination(destination);
headerAccessor.setMessageTypeIfNotSet(SimpMessageType.MESSAGE);
message = MessageBuilder.createMessage(message.getPayload(), headerAccessor.getMessageHeaders());
long timeout = this.sendTimeout;
boolean sent = (timeout >= 0)

3
spring-messaging/src/main/java/org/springframework/messaging/simp/annotation/support/PrincipalMethodArgumentResolver.java
View File

@@ -37,8 +37,7 @@ public class PrincipalMethodArgumentResolver implements HandlerMethodArgumentRes
@Override
public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
Principal user = headers.getUser();
Principal user = SimpMessageHeaderAccessor.getUser(message.getHeaders());
if (user == null) {
throw new MissingSessionUserException(message);
}

25
spring-messaging/src/main/java/org/springframework/messaging/simp/annotation/support/SendToMethodReturnValueHandler.java
View File

@@ -23,6 +23,7 @@ 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.MessageHeaders;
import org.springframework.messaging.core.MessagePostProcessor;
import org.springframework.messaging.handler.DestinationPatternsMessageCondition;
import org.springframework.messaging.handler.annotation.SendTo;
@@ -113,29 +114,28 @@ public class SendToMethodReturnValueHandler implements HandlerMethodReturnValueH
}
@Override
public void handleReturnValue(Object returnValue, MethodParameter returnType, Message<?> inputMessage)
public void handleReturnValue(Object returnValue, MethodParameter returnType, Message<?> message)
throws Exception {
if (returnValue == null) {
return;
}
SimpMessageHeaderAccessor inputHeaders = SimpMessageHeaderAccessor.wrap(inputMessage);
String sessionId = inputHeaders.getSessionId();
MessageHeaders headers = message.getHeaders();
String sessionId = SimpMessageHeaderAccessor.getSessionId(headers);
MessagePostProcessor postProcessor = new SessionHeaderPostProcessor(sessionId);
SendToUser sendToUser = returnType.getMethodAnnotation(SendToUser.class);
if (sendToUser != null) {
Principal principal = inputHeaders.getUser();
Principal principal = SimpMessageHeaderAccessor.getUser(headers);
if (principal == null) {
throw new MissingSessionUserException(inputMessage);
throw new MissingSessionUserException(message);
}
String userName = principal.getName();
if (principal instanceof DestinationUserNameProvider) {
userName = ((DestinationUserNameProvider) principal).getDestinationUserName();
}
String[] destinations = getTargetDestinations(sendToUser, inputHeaders, this.defaultUserDestinationPrefix);
String[] destinations = getTargetDestinations(sendToUser, message, this.defaultUserDestinationPrefix);
for (String destination : destinations) {
this.messagingTemplate.convertAndSendToUser(userName, destination, returnValue, postProcessor);
}
@@ -143,15 +143,14 @@ public class SendToMethodReturnValueHandler implements HandlerMethodReturnValueH
}
else {
SendTo sendTo = returnType.getMethodAnnotation(SendTo.class);
String[] destinations = getTargetDestinations(sendTo, inputHeaders, this.defaultDestinationPrefix);
String[] destinations = getTargetDestinations(sendTo, message, this.defaultDestinationPrefix);
for (String destination : destinations) {
this.messagingTemplate.convertAndSend(destination, returnValue, postProcessor);
}
}
}
protected String[] getTargetDestinations(Annotation annot, SimpMessageHeaderAccessor inputHeaders,
String defaultPrefix) {
protected String[] getTargetDestinations(Annotation annot, Message<?> message, String defaultPrefix) {
if (annot != null) {
String[] value = (String[]) AnnotationUtils.getValue(annot);
@@ -159,8 +158,8 @@ public class SendToMethodReturnValueHandler implements HandlerMethodReturnValueH
return value;
}
}
return new String[] { defaultPrefix +
inputHeaders.getHeader(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER) };
String name = DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER;
return new String[] { defaultPrefix + message.getHeaders().get(name) };
}
@@ -176,7 +175,7 @@ public class SendToMethodReturnValueHandler implements HandlerMethodReturnValueH
public Message<?> postProcessMessage(Message<?> message) {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
headers.setSessionId(this.sessionId);
return MessageBuilder.withPayload(message.getPayload()).setHeaders(headers).build();
return MessageBuilder.createMessage(message.getPayload(), headers.getMessageHeaders());
}
}

13
spring-messaging/src/main/java/org/springframework/messaging/simp/annotation/support/SimpAnnotationMethodMessageHandler.java
View File

@@ -63,6 +63,7 @@ import org.springframework.stereotype.Controller;
import org.springframework.util.AntPathMatcher;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.PathMatcher;
import org.springframework.validation.Errors;
import org.springframework.validation.Validator;
@@ -329,7 +330,7 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
@Override
protected String getDestination(Message<?> message) {
return (String) message.getHeaders().get(SimpMessageHeaderAccessor.DESTINATION_HEADER);
return (String) SimpMessageHeaderAccessor.getDestination(message.getHeaders());
}
@Override
@@ -352,13 +353,15 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
protected void handleMatch(SimpMessageMappingInfo mapping, HandlerMethod handlerMethod,
String lookupDestination, Message<?> message) {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
String matchedPattern = mapping.getDestinationConditions().getPatterns().iterator().next();
Map<String, String> vars = getPathMatcher().extractUriTemplateVariables(matchedPattern, lookupDestination);
headers.setHeader(DestinationVariableMethodArgumentResolver.DESTINATION_TEMPLATE_VARIABLES_HEADER, vars);
message = MessageBuilder.withPayload(message.getPayload()).setHeaders(headers).build();
if (!CollectionUtils.isEmpty(vars)) {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
headers.setHeader(DestinationVariableMethodArgumentResolver.DESTINATION_TEMPLATE_VARIABLES_HEADER, vars);
message = MessageBuilder.createMessage(message.getPayload(), headers.getMessageHeaders());
}
super.handleMatch(mapping, handlerMethod, lookupDestination, message);
}

22
spring-messaging/src/main/java/org/springframework/messaging/simp/annotation/support/SubscriptionMethodReturnValueHandler.java
View File

@@ -18,6 +18,7 @@ package org.springframework.messaging.simp.annotation.support;
import org.springframework.core.MethodParameter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.core.MessagePostProcessor;
import org.springframework.messaging.core.MessageSendingOperations;
import org.springframework.messaging.handler.annotation.SendTo;
@@ -71,13 +72,12 @@ public class SubscriptionMethodReturnValueHandler implements HandlerMethodReturn
return;
}
SimpMessageHeaderAccessor inputHeaders = SimpMessageHeaderAccessor.wrap(message);
String sessionId = inputHeaders.getSessionId();
String subscriptionId = inputHeaders.getSubscriptionId();
String destination = inputHeaders.getDestination();
MessageHeaders headers = message.getHeaders();
String sessionId = SimpMessageHeaderAccessor.getSessionId(headers);
String subscriptionId = SimpMessageHeaderAccessor.getSubscriptionId(headers);
String destination = SimpMessageHeaderAccessor.getDestination(headers);
Assert.state(inputHeaders.getSubscriptionId() != null,
"No subsriptiondId in input message to method " + returnType.getMethod());
Assert.state(subscriptionId != null, "No subsriptiondId in input message to method " + returnType.getMethod());
MessagePostProcessor postProcessor = new SubscriptionHeaderPostProcessor(sessionId, subscriptionId);
this.messagingTemplate.convertAndSend(destination, returnValue, postProcessor);
@@ -98,11 +98,11 @@ public class SubscriptionMethodReturnValueHandler implements HandlerMethodReturn
@Override
public Message<?> postProcessMessage(Message<?> message) {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
headers.setSessionId(this.sessionId);
headers.setSubscriptionId(this.subscriptionId);
headers.setMessageTypeIfNotSet(SimpMessageType.MESSAGE);
return MessageBuilder.withPayload(message.getPayload()).setHeaders(headers).build();
SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.wrap(message);
headerAccessor.setSessionId(this.sessionId);
headerAccessor.setSubscriptionId(this.subscriptionId);
headerAccessor.setMessageTypeIfNotSet(SimpMessageType.MESSAGE);
return MessageBuilder.createMessage(message.getPayload(), headerAccessor.getMessageHeaders());
}
}
}

42
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/AbstractSubscriptionRegistry.java
View File

@@ -19,8 +19,10 @@ package org.springframework.messaging.simp.broker;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.MultiValueMap;
/**
@@ -38,29 +40,31 @@ public abstract class AbstractSubscriptionRegistry implements SubscriptionRegist
@Override
public final void registerSubscription(Message<?> message) {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
if (!SimpMessageType.SUBSCRIBE.equals(headers.getMessageType())) {
MessageHeaders headers = message.getHeaders();
SimpMessageType type = SimpMessageHeaderAccessor.getMessageType(headers);
if (!SimpMessageType.SUBSCRIBE.equals(type)) {
logger.error("Expected SUBSCRIBE message: " + message);
return;
}
String sessionId = headers.getSessionId();
String sessionId = SimpMessageHeaderAccessor.getSessionId(headers);
if (sessionId == null) {
logger.error("Ignoring subscription. No sessionId in message: " + message);
return;
}
String subscriptionId = headers.getSubscriptionId();
String subscriptionId = SimpMessageHeaderAccessor.getSubscriptionId(headers);
if (subscriptionId == null) {
logger.error("Ignoring subscription. No subscriptionId in message: " + message);
return;
}
String destination = headers.getDestination();
String destination = SimpMessageHeaderAccessor.getDestination(headers);
if (destination == null) {
logger.error("Ignoring destination. No destination in message: " + message);
return;
}
if (logger.isDebugEnabled()) {
logger.debug("Adding subscription id=" + headers.getSubscriptionId()
+ ", destination=" + headers.getDestination());
logger.debug("Adding subscription id=" + subscriptionId + ", destination=" + destination);
}
addSubscriptionInternal(sessionId, subscriptionId, destination, message);
}
@@ -70,17 +74,20 @@ public abstract class AbstractSubscriptionRegistry implements SubscriptionRegist
@Override
public final void unregisterSubscription(Message<?> message) {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
if (!SimpMessageType.UNSUBSCRIBE.equals(headers.getMessageType())) {
MessageHeaders headers = message.getHeaders();
SimpMessageType type = SimpMessageHeaderAccessor.getMessageType(headers);
if (!SimpMessageType.UNSUBSCRIBE.equals(type)) {
logger.error("Expected UNSUBSCRIBE message: " + message);
return;
}
String sessionId = headers.getSessionId();
String sessionId = SimpMessageHeaderAccessor.getSessionId(headers);
if (sessionId == null) {
logger.error("Ignoring subscription. No sessionId in message: " + message);
return;
}
String subscriptionId = headers.getSubscriptionId();
String subscriptionId = SimpMessageHeaderAccessor.getSubscriptionId(headers);
if (subscriptionId == null) {
logger.error("Ignoring subscription. No subscriptionId in message: " + message);
return;
@@ -98,19 +105,22 @@ public abstract class AbstractSubscriptionRegistry implements SubscriptionRegist
@Override
public final MultiValueMap<String, String> findSubscriptions(Message<?> message) {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
if (!SimpMessageType.MESSAGE.equals(headers.getMessageType())) {
logger.trace("Ignoring message type " + headers.getMessageType());
MessageHeaders headers = message.getHeaders();
SimpMessageType type = SimpMessageHeaderAccessor.getMessageType(headers);
if (!SimpMessageType.MESSAGE.equals(type)) {
logger.trace("Ignoring message type " + type);
return null;
}
String destination = headers.getDestination();
String destination = SimpMessageHeaderAccessor.getDestination(headers);
if (destination == null) {
logger.trace("Ignoring message, no destination");
return null;
}
MultiValueMap<String, String> result = findSubscriptionsInternal(destination, message);
if (logger.isTraceEnabled()) {
logger.trace("Found " + result.size() + " subscriptions for destination=" + headers.getDestination());
logger.trace("Found " + result.size() + " subscriptions for destination=" + destination);
}
return result;
}

47
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/SimpleBrokerMessageHandler.java
View File

@@ -20,6 +20,7 @@ import java.util.Collection;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
@@ -111,9 +112,10 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
@Override
protected void handleMessageInternal(Message<?> message) {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
SimpMessageType messageType = headers.getMessageType();
String destination = headers.getDestination();
MessageHeaders headers = message.getHeaders();
SimpMessageType messageType = SimpMessageHeaderAccessor.getMessageType(headers);
String destination = SimpMessageHeaderAccessor.getDestination(headers);
String sessionId = SimpMessageHeaderAccessor.getSessionId(headers);
if (!checkDestinationPrefix(destination)) {
if (logger.isTraceEnabled()) {
@@ -122,27 +124,30 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
return;
}
if (SimpMessageType.SUBSCRIBE.equals(messageType)) {
if (SimpMessageType.MESSAGE.equals(messageType)) {
sendMessageToSubscribers(destination, message);
}
else if (SimpMessageType.SUBSCRIBE.equals(messageType)) {
this.subscriptionRegistry.registerSubscription(message);
}
else if (SimpMessageType.UNSUBSCRIBE.equals(messageType)) {
this.subscriptionRegistry.unregisterSubscription(message);
}
else if (SimpMessageType.MESSAGE.equals(messageType)) {
sendMessageToSubscribers(headers.getDestination(), message);
}
else if (SimpMessageType.DISCONNECT.equals(messageType)) {
String sessionId = headers.getSessionId();
this.subscriptionRegistry.unregisterAllSubscriptions(sessionId);
this.subscriptionRegistry.unregisterAllSubscriptions(sessionId);
}
else if (SimpMessageType.CONNECT.equals(messageType)) {
SimpMessageHeaderAccessor replyHeaders = SimpMessageHeaderAccessor.create(SimpMessageType.CONNECT_ACK);
replyHeaders.setSessionId(headers.getSessionId());
replyHeaders.setHeader(SimpMessageHeaderAccessor.CONNECT_MESSAGE_HEADER, message);
Message<byte[]> connectAck = MessageBuilder.withPayload(EMPTY_PAYLOAD).setHeaders(replyHeaders).build();
SimpMessageHeaderAccessor accessor = SimpMessageHeaderAccessor.create(SimpMessageType.CONNECT_ACK);
accessor.setSessionId(sessionId);
accessor.setHeader(SimpMessageHeaderAccessor.CONNECT_MESSAGE_HEADER, message);
Message<byte[]> connectAck = MessageBuilder.createMessage(EMPTY_PAYLOAD, accessor.getMessageHeaders());
this.clientOutboundChannel.send(connectAck);
}
else {
if (logger.isTraceEnabled()) {
logger.trace("Message type not supported. Ignoring: " + message);
}
}
}
protected void sendMessageToSubscribers(String destination, Message<?> message) {
@@ -153,17 +158,17 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
}
for (String sessionId : subscriptions.keySet()) {
for (String subscriptionId : subscriptions.get(sessionId)) {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
headers.setSessionId(sessionId);
headers.setSubscriptionId(subscriptionId);
SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.create(SimpMessageType.MESSAGE);
headerAccessor.setSessionId(sessionId);
headerAccessor.setSubscriptionId(subscriptionId);
headerAccessor.copyHeadersIfAbsent(message.getHeaders());
Object payload = message.getPayload();
Message<?> clientMessage = MessageBuilder.withPayload(payload).setHeaders(headers).build();
Message<?> reply = MessageBuilder.createMessage(payload, headerAccessor.getMessageHeaders());
try {
this.clientOutboundChannel.send(clientMessage);
this.clientOutboundChannel.send(reply);
}
catch (Throwable ex) {
logger.error("Failed to send message to destination=" + destination +
", sessionId=" + sessionId + ", subscriptionId=" + subscriptionId, ex);
logger.error("Failed to send message=" + message, ex);
}
}
}

110
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java
View File

@@ -20,7 +20,6 @@ import java.util.Collection;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -30,6 +29,7 @@ import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.simp.broker.AbstractBrokerMessageHandler;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.messaging.tcp.FixedIntervalReconnectStrategy;
import org.springframework.messaging.tcp.TcpConnection;
import org.springframework.messaging.tcp.TcpConnectionHandler;
@@ -79,9 +79,9 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
private static final Message<byte[]> HEARTBEAT_MESSAGE;
static {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.create(SimpMessageType.HEARTBEAT);
HEARTBEAT_MESSAGE = MessageBuilder.withPayload(new byte[] {'\n'}).setHeaders(headers).build();
EMPTY_TASK.run();
StompHeaderAccessor accessor = StompHeaderAccessor.createForHeartbeat();
HEARTBEAT_MESSAGE = MessageBuilder.createMessage(StompDecoder.HEARTBEAT_PAYLOAD, accessor.getMessageHeaders());
}
@@ -370,37 +370,53 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
@Override
protected void handleMessageInternal(Message<?> message) {
StompHeaderAccessor headers = StompHeaderAccessor.wrap(message);
String sessionId = headers.getSessionId();
String sessionId = SimpMessageHeaderAccessor.getSessionId(message.getHeaders());
if (!isBrokerAvailable()) {
if (sessionId == null || sessionId == SystemStompConnectionHandler.SESSION_ID) {
if (sessionId == null || SystemStompConnectionHandler.SESSION_ID.equals(sessionId)) {
throw new MessageDeliveryException("Message broker is not active.");
}
if (logger.isTraceEnabled()) {
logger.trace("Message broker is not active. Ignoring message id=" + message.getHeaders().getId());
logger.trace("Message broker is not active. Ignoring: " + message);
}
return;
}
String destination = headers.getDestination();
StompCommand command = headers.getCommand();
SimpMessageType messageType = headers.getMessageType();
StompHeaderAccessor stompAccessor;
StompCommand command;
if (SimpMessageType.MESSAGE.equals(messageType)) {
sessionId = (sessionId == null) ? SystemStompConnectionHandler.SESSION_ID : sessionId;
headers.setSessionId(sessionId);
command = headers.updateStompCommandAsClientMessage();
message = MessageBuilder.withPayload(message.getPayload()).setHeaders(headers).build();
MessageHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, MessageHeaderAccessor.class);
if (accessor == null) {
logger.error("No header accessor, please use SimpMessagingTemplate. Ignoring: " + message);
return;
}
else if (accessor instanceof StompHeaderAccessor) {
stompAccessor = (StompHeaderAccessor) accessor;
command = stompAccessor.getCommand();
}
else if (accessor instanceof SimpMessageHeaderAccessor) {
stompAccessor = StompHeaderAccessor.wrap(message);
command = stompAccessor.getCommand();
if (command == null) {
command = stompAccessor.updateStompCommandAsClientMessage();
}
}
else {
// Should not happen
logger.error("Unexpected header accessor type: " + accessor + ". Ignoring: " + message);
return;
}
if (sessionId == null) {
if (logger.isWarnEnabled()) {
logger.warn("No sessionId, ignoring message: " + message);
if (!SimpMessageType.MESSAGE.equals(stompAccessor.getMessageType())) {
logger.error("Only STOMP SEND frames supported on \"system\" connection. Ignoring: " + message);
return;
}
return;
sessionId = SystemStompConnectionHandler.SESSION_ID;
stompAccessor.setSessionId(sessionId);
}
String destination = stompAccessor.getDestination();
if ((command != null) && command.requiresDestination() && !checkDestinationPrefix(destination)) {
if (logger.isTraceEnabled()) {
logger.trace("Ignoring message to destination=" + destination);
@@ -412,20 +428,21 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
logger.trace("Processing message=" + message);
}
if (SimpMessageType.CONNECT.equals(messageType)) {
if (StompCommand.CONNECT.equals(command)) {
if (logger.isDebugEnabled()) {
logger.debug("Processing CONNECT (total connected=" + this.connectionHandlers.size() + ")");
}
headers.setLogin(this.clientLogin);
headers.setPasscode(this.clientPasscode);
stompAccessor = (stompAccessor.isMutable() ? stompAccessor : StompHeaderAccessor.wrap(message));
stompAccessor.setLogin(this.clientLogin);
stompAccessor.setPasscode(this.clientPasscode);
if (getVirtualHost() != null) {
headers.setHost(getVirtualHost());
stompAccessor.setHost(getVirtualHost());
}
StompConnectionHandler handler = new StompConnectionHandler(sessionId, headers);
StompConnectionHandler handler = new StompConnectionHandler(sessionId, stompAccessor);
this.connectionHandlers.put(sessionId, handler);
this.tcpClient.connect(handler);
}
else if (SimpMessageType.DISCONNECT.equals(messageType)) {
else if (StompCommand.DISCONNECT.equals(command)) {
StompConnectionHandler handler = this.connectionHandlers.get(sessionId);
if (handler == null) {
if (logger.isTraceEnabled()) {
@@ -433,7 +450,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
}
return;
}
handler.forward(message);
handler.forward(message, stompAccessor);
}
else {
StompConnectionHandler handler = this.connectionHandlers.get(sessionId);
@@ -443,7 +460,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
}
return;
}
handler.forward(message);
handler.forward(message, stompAccessor);
}
}
@@ -486,7 +503,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
logger.debug("Established TCP connection to broker in session '" + this.sessionId + "'");
}
this.tcpConnection = connection;
connection.send(MessageBuilder.withPayload(EMPTY_PAYLOAD).setHeaders(this.connectHeaders).build());
connection.send(MessageBuilder.createMessage(EMPTY_PAYLOAD, this.connectHeaders.getMessageHeaders()));
}
@Override
@@ -522,7 +539,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.ERROR);
headers.setSessionId(this.sessionId);
headers.setMessage(errorText);
Message<?> errorMessage = MessageBuilder.withPayload(new byte[0]).setHeaders(headers).build();
Message<?> errorMessage = MessageBuilder.createMessage(EMPTY_PAYLOAD, headers.getMessageHeaders());
sendMessageToClient(errorMessage);
}
}
@@ -536,20 +553,23 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
@Override
public void handleMessage(Message<byte[]> message) {
StompHeaderAccessor headers = StompHeaderAccessor.wrap(message);
if (SimpMessageType.HEARTBEAT.equals(headers.getMessageType())) {
StompHeaderAccessor headerAccessor =
MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
if (headerAccessor.isHeartbeat()) {
logger.trace("Received broker heartbeat");
}
else if (logger.isDebugEnabled()) {
logger.debug("Received message from broker in session '" + this.sessionId + "'");
}
if (StompCommand.CONNECTED == headers.getCommand()) {
afterStompConnected(headers);
if (StompCommand.CONNECTED == headerAccessor.getCommand()) {
afterStompConnected(headerAccessor);
}
headers.setSessionId(this.sessionId);
message = MessageBuilder.withPayload(message.getPayload()).setHeaders(headers).build();
headerAccessor.setSessionId(this.sessionId);
headerAccessor.setImmutable();
sendMessageToClient(message);
}
@@ -630,9 +650,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
clearConnection();
}
catch (Throwable t) {
if (logger.isErrorEnabled()) {
// Ignore
}
// Ignore
}
}
}
@@ -661,7 +679,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
* @return a future to wait for the result
*/
@SuppressWarnings("unchecked")
public ListenableFuture<Void> forward(final Message<?> message) {
public ListenableFuture<Void> forward(Message<?> message, final StompHeaderAccessor headerAccessor) {
TcpConnection<byte[]> conn = this.tcpConnection;
@@ -682,8 +700,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
}
if (logger.isDebugEnabled()) {
StompHeaderAccessor headers = StompHeaderAccessor.wrap(message);
if (SimpMessageType.HEARTBEAT.equals(headers.getMessageType())) {
if (headerAccessor.isHeartbeat()) {
logger.trace("Forwarding heartbeat to broker");
}
else {
@@ -691,13 +708,16 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
}
}
if (headerAccessor.isMutable() && headerAccessor.isModified()) {
message = MessageBuilder.createMessage(message.getPayload(), headerAccessor.getMessageHeaders());
}
ListenableFuture<Void> future = conn.send((Message<byte[]>) message);
future.addCallback(new ListenableFutureCallback<Void>() {
@Override
public void onSuccess(Void result) {
StompCommand command = StompHeaderAccessor.wrap(message).getCommand();
if (command == StompCommand.DISCONNECT) {
if (headerAccessor.getCommand() == StompCommand.DISCONNECT) {
clearConnection();
}
}
@@ -707,7 +727,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
// already reset
}
else {
handleTcpConnectionFailure("Failed to send message " + message, t);
handleTcpConnectionFailure("Failed to send message " + headerAccessor, t);
}
}
});
@@ -777,9 +797,9 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
}
@Override
public ListenableFuture<Void> forward(Message<?> message) {
public ListenableFuture<Void> forward(Message<?> message, StompHeaderAccessor headerAccessor) {
try {
ListenableFuture<Void> future = super.forward(message);
ListenableFuture<Void> future = super.forward(message, headerAccessor);
future.get();
return future;
}

26
spring-messaging/src/main/java/org/springframework/messaging/simp/user/DefaultUserDestinationResolver.java
View File

@@ -19,6 +19,7 @@ package org.springframework.messaging.simp.user;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.util.Assert;
@@ -100,34 +101,34 @@ public class DefaultUserDestinationResolver implements UserDestinationResolver {
@Override
public UserDestinationResult resolveDestination(Message<?> message) {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
DestinationInfo info = parseUserDestination(headers);
String destination = SimpMessageHeaderAccessor.getDestination(message.getHeaders());
DestinationInfo info = parseUserDestination(message);
if (info == null) {
return null;
}
Set<String> targetDestinations = new HashSet<String>();
for (String sessionId : info.getSessionIds()) {
targetDestinations.add(getTargetDestination(
headers.getDestination(), info.getDestinationWithoutPrefix(), sessionId, info.getUser()));
targetDestinations.add(getTargetDestination(destination,
info.getDestinationWithoutPrefix(), sessionId, info.getUser()));
}
return new UserDestinationResult(headers.getDestination(),
return new UserDestinationResult(destination,
targetDestinations, info.getSubscribeDestination(), info.getUser());
}
private DestinationInfo parseUserDestination(SimpMessageHeaderAccessor headers) {
private DestinationInfo parseUserDestination(Message<?> message) {
String destination = headers.getDestination();
MessageHeaders headers = message.getHeaders();
SimpMessageType messageType = SimpMessageHeaderAccessor.getMessageType(headers);
String destination = SimpMessageHeaderAccessor.getDestination(headers);
Principal principal = SimpMessageHeaderAccessor.getUser(headers);
String destinationWithoutPrefix;
String subscribeDestination;
String user;
Set<String> sessionIds;
Principal principal = headers.getUser();
SimpMessageType messageType = headers.getMessageType();
if (SimpMessageType.SUBSCRIBE.equals(messageType) || SimpMessageType.UNSUBSCRIBE.equals(messageType)) {
if (!checkDestination(destination, this.destinationPrefix)) {
return null;
@@ -136,14 +137,15 @@ public class DefaultUserDestinationResolver implements UserDestinationResolver {
logger.error("Ignoring message, no principal info available");
return null;
}
if (headers.getSessionId() == null) {
String sessionId = SimpMessageHeaderAccessor.getSessionId(headers);
if (sessionId == null) {
logger.error("Ignoring message, no session id available");
return null;
}
destinationWithoutPrefix = destination.substring(this.destinationPrefix.length()-1);
subscribeDestination = destination;
user = principal.getName();
sessionIds = Collections.singleton(headers.getSessionId());
sessionIds = Collections.singleton(sessionId);
}
else if (SimpMessageType.MESSAGE.equals(messageType)) {
if (!checkDestination(destination, this.destinationPrefix)) {

12
spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationMessageHandler.java
View File

@@ -152,16 +152,16 @@ public class UserDestinationMessageHandler implements MessageHandler, SmartLifec
if (destinations.isEmpty()) {
return;
}
SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.wrap(message);
if (SimpMessageType.MESSAGE.equals(headerAccessor.getMessageType())) {
if (SimpMessageType.MESSAGE.equals(SimpMessageHeaderAccessor.getMessageType(message.getHeaders()))) {
SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.wrap(message);
headerAccessor.setHeader(SimpMessageHeaderAccessor.ORIGINAL_DESTINATION, result.getSubscribeDestination());
message = MessageBuilder.withPayload(message.getPayload()).setHeaders(headerAccessor).build();
message = MessageBuilder.createMessage(message.getPayload(), headerAccessor.getMessageHeaders());
}
for (String targetDestination : destinations) {
for (String destination : destinations) {
if (logger.isDebugEnabled()) {
logger.debug("Sending message to resolved destination=" + targetDestination);
logger.debug("Sending message to resolved destination=" + destination);
}
this.brokerMessagingTemplate.send(targetDestination, message);
this.brokerMessagingTemplate.send(destination, message);
}
}