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

STOMP and WebSocket messaging related logging updates

Browse Source 
This change removes most logging at INFO level and also ensures the
amount of information logged at DEBUG level is useful, brief, and
not duplicated.

Also added is custom logging for STOMP frames to ensure very readable
and consise output.

Issue: SPR-11934
This commit is contained in:
Rossen Stoyanchev
2014-07-09 00:30:22 -04:00
parent ab4864da2a
commit 48236be4a2
65 changed files with 763 additions and 425 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-messaging/src/main/java/org/springframework/messaging/handler/HandlerMethod.java
View File

@@ -21,7 +21,6 @@ import java.lang.reflect.Method;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.BridgeMethodResolver;
import org.springframework.core.MethodParameter;
@@ -44,7 +43,6 @@ import org.springframework.util.ClassUtils;
*/
public class HandlerMethod {
/** Logger that is available to subclasses */
protected final Log logger = LogFactory.getLog(HandlerMethod.class);
private final Object bean;
@@ -238,6 +236,11 @@ public class HandlerMethod {
return this.bean.hashCode() * 31 + this.method.hashCode();
}
public String getShortLogMessage() {
int args = method.getParameterTypes().length;
return getBeanType().getName() + "#" + this.method.getName() + "[" + args + " args]";
}
@Override
public String toString() {
return this.method.toGenericString();

3
spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/support/HeaderMethodArgumentResolver.java
View File

@@ -80,9 +80,6 @@ public class HeaderMethodArgumentResolver extends AbstractNamedValueMethodArgume
if (name.startsWith("nativeHeaders.")) {
name = name.substring("nativeHeaders.".length());
if (logger.isDebugEnabled()) {
logger.debug("Looking up native header '" + name + "'");
}
}
if ((nativeHeaders == null) || !nativeHeaders.containsKey(name)) {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -322,26 +322,23 @@ public abstract class AbstractMethodMessageHandler<T>
@Override
public void handleMessage(Message<?> message) throws MessagingException {
String destination = getDestination(message);
if (destination == null) {
return;
}
String lookupDestination = getLookupDestination(destination);
if (lookupDestination == null) {
return;
}
if (logger.isDebugEnabled()) {
logger.debug("Handling message to " + destination);
}
MessageHeaderAccessor headerAccessor = MessageHeaderAccessor.getMutableAccessor(message);
headerAccessor.setHeader(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER, lookupDestination);
headerAccessor.setLeaveMutable(true);
message = MessageBuilder.createMessage(message.getPayload(), headerAccessor.getMessageHeaders());
if (logger.isDebugEnabled()) {
logger.debug("Searching methods to handle " + headerAccessor.getShortLogMessage(message.getPayload()));
}
handleMessageInternal(message, lookupDestination);
headerAccessor.setImmutable();
}
@@ -377,24 +374,20 @@ public abstract class AbstractMethodMessageHandler<T>
if (mappingsByUrl != null) {
addMatchesToCollection(mappingsByUrl, message, matches);
}
if (matches.isEmpty()) {
// No direct hits, go through all mappings
Set<T> allMappings = this.handlerMethods.keySet();
addMatchesToCollection(allMappings, message, matches);
}
if (matches.isEmpty()) {
handleNoMatch(handlerMethods.keySet(), lookupDestination, message);
return;
}
Comparator<Match> comparator = new MatchComparator(getMappingComparator(message));
Collections.sort(matches, comparator);
if (logger.isTraceEnabled()) {
logger.trace("Found " + matches.size() + " matching mapping(s) for [" +
lookupDestination + "] : " + matches);
logger.trace("Found " + matches.size() + " methods: " + matches);
}
Match bestMatch = matches.get(0);
@@ -440,18 +433,14 @@ public abstract class AbstractMethodMessageHandler<T>
protected void handleMatch(T mapping, HandlerMethod handlerMethod, String lookupDestination, Message<?> message) {
if (logger.isDebugEnabled()) {
logger.debug("Message matched to " + handlerMethod);
logger.debug("Invoking " + handlerMethod.getShortLogMessage());
}
handlerMethod = handlerMethod.createWithResolvedBean();
InvocableHandlerMethod invocable = new InvocableHandlerMethod(handlerMethod);
invocable.setMessageMethodArgumentResolvers(this.argumentResolvers);
try {
Object returnValue = invocable.invoke(message);
MethodParameter returnType = handlerMethod.getReturnType();
if (void.class.equals(returnType.getParameterType())) {
return;
@@ -467,26 +456,27 @@ public abstract class AbstractMethodMessageHandler<T>
}
protected void processHandlerMethodException(HandlerMethod handlerMethod, Exception ex, Message<?> message) {
if (logger.isDebugEnabled()) {
logger.debug("Searching methods to handle " + ex.getClass().getSimpleName());
}
Class<?> beanType = handlerMethod.getBeanType();
AbstractExceptionHandlerMethodResolver resolver = this.exceptionHandlerCache.get(beanType);
if (resolver == null) {
resolver = createExceptionHandlerMethodResolverFor(beanType);
this.exceptionHandlerCache.put(beanType, resolver);
}
Method method = resolver.resolveMethod(ex);
if (method == null) {
logger.error("Unhandled exception", ex);
return;
}
InvocableHandlerMethod invocable = new InvocableHandlerMethod(handlerMethod.getBean(), method);
invocable.setMessageMethodArgumentResolvers(this.argumentResolvers);
if (logger.isDebugEnabled()) {
logger.debug("Invoking " + invocable.getShortLogMessage());
}
try {
Object returnValue = invocable.invoke(message, ex);
MethodParameter returnType = invocable.getReturnType();
if (void.class.equals(returnType.getParameterType())) {
return;
@@ -497,17 +487,21 @@ public abstract class AbstractMethodMessageHandler<T>
logger.error("Error while handling exception", t);
return;
}
}
protected abstract AbstractExceptionHandlerMethodResolver createExceptionHandlerMethodResolverFor(Class<?> beanType);
protected void handleNoMatch(Set<T> ts, String lookupDestination, Message<?> message) {
if (logger.isDebugEnabled()) {
logger.debug("No matching method found.");
logger.debug("No matching methods.");
}
}
@Override
public String toString() {
return getClass().getSimpleName() + "[prefixes=" + getDestinationPrefixes() + "]";
}
/**
* A thin wrapper around a matched HandlerMethod and its matched mapping for

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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.
@@ -22,8 +22,6 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
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;
@@ -38,8 +36,6 @@ import org.springframework.util.Assert;
*/
public class HandlerMethodArgumentResolverComposite implements HandlerMethodArgumentResolver {
protected final Log logger = LogFactory.getLog(getClass());
private final List<HandlerMethodArgumentResolver> argumentResolvers = new LinkedList<HandlerMethodArgumentResolver>();
private final Map<MethodParameter, HandlerMethodArgumentResolver> argumentResolverCache =

8
spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/HandlerMethodReturnValueHandlerComposite.java
View File

@@ -27,6 +27,8 @@ import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* A HandlerMethodReturnValueHandler that wraps and delegates to others.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
@@ -79,9 +81,6 @@ 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;
}
}
@@ -94,6 +93,9 @@ public class HandlerMethodReturnValueHandlerComposite implements HandlerMethodRe
HandlerMethodReturnValueHandler handler = getReturnValueHandler(returnType);
Assert.notNull(handler, "No handler for return value type [" + returnType.getParameterType().getName() + "]");
if (logger.isTraceEnabled()) {
logger.trace("Processing return value with " + handler);
}
handler.handleReturnValue(returnValue, returnType, message);
}

12
spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/InvocableHandlerMethod.java
View File

@@ -96,15 +96,11 @@ public class InvocableHandlerMethod extends HandlerMethod {
public final Object invoke(Message<?> message, Object... providedArgs) throws Exception {
Object[] args = getMethodArgumentValues(message, providedArgs);
if (logger.isTraceEnabled()) {
StringBuilder sb = new StringBuilder("Invoking [");
sb.append(getBeanType().getSimpleName()).append(".");
sb.append(getMethod().getName()).append("] method with arguments ");
sb.append(Arrays.asList(args));
logger.trace(sb.toString());
logger.trace("Resolved arguments: " + Arrays.asList(args));
}
Object returnValue = invoke(args);
if (logger.isTraceEnabled()) {
logger.trace("Method [" + getMethod().getName() + "] returned [" + returnValue + "]");
logger.trace("Returned value: " + returnValue);
}
return returnValue;
}
@@ -136,8 +132,8 @@ public class InvocableHandlerMethod extends HandlerMethod {
}
}
if (args[i] == null) {
String msg = getArgumentResolutionErrorMessage("No suitable resolver for argument", i);
throw new IllegalStateException(msg);
String error = getArgumentResolutionErrorMessage("No suitable resolver for argument", i);
throw new IllegalStateException(error);
}
}
return args;

8
spring-messaging/src/main/java/org/springframework/messaging/simp/SimpAttributes.java
View File

@@ -72,9 +72,11 @@ public class SimpAttributes {
MessageHeaders headers = message.getHeaders();
String sessionId = SimpMessageHeaderAccessor.getSessionId(headers);
Map<String, Object> sessionAttributes = SimpMessageHeaderAccessor.getSessionAttributes(headers);
if (sessionId == null || sessionAttributes == null) {
throw new IllegalStateException(
"Message does not contain SiMP session id or attributes: " + message);
if (sessionId == null) {
throw new IllegalStateException("No session id in " + message);
}
if (sessionAttributes == null) {
throw new IllegalStateException("No session attributes in " + message);
}
return new SimpAttributes(sessionId, sessionAttributes);
}

48
spring-messaging/src/main/java/org/springframework/messaging/simp/SimpMessageHeaderAccessor.java
View File

@@ -20,11 +20,13 @@ import java.security.Principal;
import java.util.List;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.IdTimestampMessageHeaderInitializer;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.messaging.support.NativeMessageHeaderAccessor;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
/**
* A base class for working with message headers in simple messaging protocols that
@@ -241,4 +243,50 @@ public class SimpMessageHeaderAccessor extends NativeMessageHeaderAccessor {
return (Principal) headers.get(USER_HEADER);
}
@Override
public String getShortLogMessage(Object payload) {
if (getMessageType() == null) {
return super.getDetailedLogMessage(payload);
}
StringBuilder sb = getBaseLogMessage();
if (!CollectionUtils.isEmpty(getSessionAttributes())) {
sb.append(" attributes[").append(getSessionAttributes().size()).append("]");
}
sb.append(getShortPayloadLogMessage(payload));
return sb.toString();
}
@SuppressWarnings("unchecked")
@Override
public String getDetailedLogMessage(Object payload) {
if (getMessageType() == null) {
return super.getDetailedLogMessage(payload);
}
StringBuilder sb = getBaseLogMessage();
if (!CollectionUtils.isEmpty(getSessionAttributes())) {
sb.append(" attributes=").append(getSessionAttributes());
}
if (!CollectionUtils.isEmpty((Map<String, List<String>>) getHeader(NATIVE_HEADERS))) {
sb.append(" nativeHeaders=").append((Map<String, List<String>>) getHeader(NATIVE_HEADERS));
}
sb.append(getDetailedPayloadLogMessage(payload));
return sb.toString();
}
private StringBuilder getBaseLogMessage() {
StringBuilder sb = new StringBuilder();
sb.append(getMessageType().name());
if (getDestination() != null) {
sb.append(" destination=").append(getDestination());
}
if (getSubscriptionId() != null) {
sb.append(" subscriptionId=").append(getSubscriptionId());
}
sb.append(" session=").append(getSessionId());
if (getUser() != null) {
sb.append(" user=").append(getUser().getName());
}
return sb;
}
}

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

@@ -137,7 +137,6 @@ public class SendToMethodReturnValueHandler implements HandlerMethodReturnValueH
if (returnValue == null) {
return;
}
MessageHeaders headers = message.getHeaders();
String sessionId = SimpMessageHeaderAccessor.getSessionId(headers);
@@ -161,7 +160,6 @@ public class SendToMethodReturnValueHandler implements HandlerMethodReturnValueH
this.messagingTemplate.convertAndSendToUser(user, destination, returnValue, createHeaders(sessionId));
}
}
return;
}
else {
SendTo sendTo = returnType.getMethodAnnotation(SendTo.class);
@@ -182,7 +180,6 @@ public class SendToMethodReturnValueHandler implements HandlerMethodReturnValueH
}
protected String[] getTargetDestinations(Annotation annotation, Message<?> message, String defaultPrefix) {
if (annotation != null) {
String[] value = (String[]) AnnotationUtils.getValue(annotation);
if (!ObjectUtils.isEmpty(value)) {

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

@@ -16,6 +16,8 @@
package org.springframework.messaging.simp.annotation.support;
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.messaging.MessageHeaders;
@@ -45,6 +47,9 @@ import org.springframework.util.Assert;
*/
public class SubscriptionMethodReturnValueHandler implements HandlerMethodReturnValueHandler {
private static Log logger = LogFactory.getLog(SubscriptionMethodReturnValueHandler.class);
private final MessageSendingOperations<String> messagingTemplate;
private MessageHeaderInitializer headerInitializer;
@@ -87,13 +92,10 @@ public class SubscriptionMethodReturnValueHandler implements HandlerMethodReturn
}
@Override
public void handleReturnValue(Object returnValue, MethodParameter returnType, Message<?> message)
throws Exception {
public void handleReturnValue(Object returnValue, MethodParameter returnType, Message<?> message) throws Exception {
if (returnValue == null) {
return;
}
MessageHeaders headers = message.getHeaders();
String destination = SimpMessageHeaderAccessor.getDestination(headers);
String sessionId = SimpMessageHeaderAccessor.getSessionId(headers);
@@ -102,6 +104,10 @@ public class SubscriptionMethodReturnValueHandler implements HandlerMethodReturn
Assert.state(subscriptionId != null,
"No subscriptionId in message=" + message + ", method=" + returnType.getMethod());
if (logger.isDebugEnabled()) {
logger.debug("Reply to @SubscribeMapping: " + returnValue);
}
this.messagingTemplate.convertAndSend(destination, returnValue, createHeaders(sessionId, subscriptionId));
}

24
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/AbstractBrokerMessageHandler.java
View File

@@ -130,15 +130,15 @@ public abstract class AbstractBrokerMessageHandler
}
@Override
public final void start() {
public void start() {
synchronized (this.lifecycleMonitor) {
if (logger.isDebugEnabled()) {
logger.debug("Starting...");
if (logger.isInfoEnabled()) {
logger.info("Starting...");
}
startInternal();
this.running = true;
if (logger.isDebugEnabled()) {
logger.debug("Started.");
if (logger.isInfoEnabled()) {
logger.info("Started.");
}
}
}
@@ -147,15 +147,15 @@ public abstract class AbstractBrokerMessageHandler
}
@Override
public final void stop() {
public void stop() {
synchronized (this.lifecycleMonitor) {
if (logger.isDebugEnabled()) {
logger.debug("Stopping...");
if (logger.isInfoEnabled()) {
logger.info("Stopping...");
}
stopInternal();
this.running = false;
if (logger.isDebugEnabled()) {
logger.debug("Stopped.");
logger.info("Stopped.");
}
}
}
@@ -172,7 +172,7 @@ public abstract class AbstractBrokerMessageHandler
}
@Override
public final void handleMessage(Message<?> message) {
public void handleMessage(Message<?> message) {
if (!this.running) {
if (logger.isTraceEnabled()) {
logger.trace(this + " not running yet. Ignoring " + message);
@@ -200,7 +200,7 @@ public abstract class AbstractBrokerMessageHandler
boolean shouldPublish = this.brokerAvailable.compareAndSet(false, true);
if (this.eventPublisher != null && shouldPublish) {
if (logger.isInfoEnabled()) {
logger.info("Publishing " + this.availableEvent);
logger.info(this.availableEvent);
}
this.eventPublisher.publishEvent(this.availableEvent);
}
@@ -210,7 +210,7 @@ public abstract class AbstractBrokerMessageHandler
boolean shouldPublish = this.brokerAvailable.compareAndSet(true, false);
if (this.eventPublisher != null && shouldPublish) {
if (logger.isInfoEnabled()) {
logger.info("Publishing " + this.notAvailableEvent);
logger.info(this.notAvailableEvent);
}
this.eventPublisher.publishEvent(this.notAvailableEvent);
}

2
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/DefaultSubscriptionRegistry.java
View File

@@ -275,7 +275,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
@Override
public String toString() {
return "registry[" + sessions.size() + " session(s)]";
return "registry[" + sessions.size() + " sessions]";
}
}

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

@@ -25,6 +25,7 @@ import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.messaging.support.MessageHeaderInitializer;
import org.springframework.util.Assert;
import org.springframework.util.MultiValueMap;
@@ -138,35 +139,44 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
return;
}
SimpMessageHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, SimpMessageHeaderAccessor.class);
if (accessor == null) {
throw new IllegalStateException(
"No header accessor (not using the SimpMessagingTemplate?): " + message);
}
if (SimpMessageType.MESSAGE.equals(messageType)) {
if (logger.isDebugEnabled()) {
logger.debug("Processing " + accessor.getShortLogMessage(message.getPayload()));
}
sendMessageToSubscribers(destination, message);
}
else if (SimpMessageType.CONNECT.equals(messageType)) {
if (logger.isInfoEnabled()) {
logger.info("Handling CONNECT: " + message);
if (logger.isDebugEnabled()) {
logger.debug("Processing " + accessor.getShortLogMessage(EMPTY_PAYLOAD));
}
SimpMessageHeaderAccessor accessor = SimpMessageHeaderAccessor.create(SimpMessageType.CONNECT_ACK);
initHeaders(accessor);
accessor.setSessionId(sessionId);
accessor.setHeader(SimpMessageHeaderAccessor.CONNECT_MESSAGE_HEADER, message);
Message<byte[]> connectAck = MessageBuilder.createMessage(EMPTY_PAYLOAD, accessor.getMessageHeaders());
this.clientOutboundChannel.send(connectAck);
SimpMessageHeaderAccessor connectAck = SimpMessageHeaderAccessor.create(SimpMessageType.CONNECT_ACK);
initHeaders(connectAck);
connectAck.setSessionId(sessionId);
connectAck.setHeader(SimpMessageHeaderAccessor.CONNECT_MESSAGE_HEADER, message);
Message<byte[]> messageOut = MessageBuilder.createMessage(EMPTY_PAYLOAD, connectAck.getMessageHeaders());
this.clientOutboundChannel.send(messageOut);
}
else if (SimpMessageType.DISCONNECT.equals(messageType)) {
if (logger.isInfoEnabled()) {
logger.info("Handling DISCONNECT: " + message);
if (logger.isDebugEnabled()) {
logger.debug("Processing " + accessor.getShortLogMessage(EMPTY_PAYLOAD));
}
this.subscriptionRegistry.unregisterAllSubscriptions(sessionId);
}
else if (SimpMessageType.SUBSCRIBE.equals(messageType)) {
if (logger.isDebugEnabled()) {
logger.debug("Handling SUBSCRIBE: " + message);
logger.debug("Processing " + accessor.getShortLogMessage(EMPTY_PAYLOAD));
}
this.subscriptionRegistry.registerSubscription(message);
}
else if (SimpMessageType.UNSUBSCRIBE.equals(messageType)) {
if (logger.isDebugEnabled()) {
logger.debug("Handling UNSUBSCRIBE: " + message);
logger.debug("Processing " + accessor.getShortLogMessage(EMPTY_PAYLOAD));
}
this.subscriptionRegistry.unregisterSubscription(message);
}
@@ -180,8 +190,8 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
protected void sendMessageToSubscribers(String destination, Message<?> message) {
MultiValueMap<String,String> subscriptions = this.subscriptionRegistry.findSubscriptions(message);
if ((subscriptions.size() > 0) && logger.isTraceEnabled()) {
logger.trace("Sending to " + subscriptions.size() + " subscriber(s): " + message);
if ((subscriptions.size() > 0) && logger.isDebugEnabled()) {
logger.debug("Broadcasting to " + subscriptions.size() + " sessions.");
}
for (String sessionId : subscriptions.keySet()) {
for (String subscriptionId : subscriptions.get(sessionId)) {

9
spring-messaging/src/main/java/org/springframework/messaging/simp/config/AbstractMessageBrokerConfiguration.java
View File

@@ -357,17 +357,20 @@ public abstract class AbstractMessageBrokerConfiguration implements ApplicationC
private static final AbstractBrokerMessageHandler noopBroker = new AbstractBrokerMessageHandler() {
@Override
protected void startInternal() {
public void start() {
}
@Override
protected void stopInternal() {
public void stop() {
}
@Override
public void handleMessage(Message<?> message) {
}
@Override
protected void handleMessageInternal(Message<?> message) {
}
};
}

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

@@ -72,6 +72,8 @@ import org.springframework.util.concurrent.ListenableFutureTask;
*/
public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler {
public static final String SYSTEM_SESSION_ID = "_system_";
private static final byte[] EMPTY_PAYLOAD = new byte[0];
private static final ListenableFutureTask<Void> EMPTY_TASK = new ListenableFutureTask<Void>(new VoidCallable());
@@ -380,14 +382,18 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
logger.info("Connecting \"system\" session to " + this.relayHost + ":" + this.relayPort);
}
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.CONNECT);
headers.setAcceptVersion("1.1,1.2");
headers.setLogin(this.systemLogin);
headers.setPasscode(this.systemPasscode);
headers.setHeartbeat(this.systemHeartbeatSendInterval, this.systemHeartbeatReceiveInterval);
headers.setHost(getVirtualHost());
StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.CONNECT);
accessor.setAcceptVersion("1.1,1.2");
accessor.setLogin(this.systemLogin);
accessor.setPasscode(this.systemPasscode);
accessor.setHeartbeat(this.systemHeartbeatSendInterval, this.systemHeartbeatReceiveInterval);
accessor.setHost(getVirtualHost());
accessor.setSessionId(SYSTEM_SESSION_ID);
if (logger.isDebugEnabled()) {
logger.debug("Forwarding " + accessor.getShortLogMessage(EMPTY_PAYLOAD));
}
SystemStompConnectionHandler handler = new SystemStompConnectionHandler(headers);
SystemStompConnectionHandler handler = new SystemStompConnectionHandler(accessor);
this.connectionHandlers.put(handler.getSessionId(), handler);
this.stats.incrementConnectCount();
@@ -416,7 +422,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
String sessionId = SimpMessageHeaderAccessor.getSessionId(message.getHeaders());
if (!isBrokerAvailable()) {
if (sessionId == null || SystemStompConnectionHandler.SESSION_ID.equals(sessionId)) {
if (sessionId == null || SYSTEM_SESSION_ID.equals(sessionId)) {
throw new MessageDeliveryException("Message broker not active. Consider subscribing to " +
"receive BrokerAvailabilityEvent's from an ApplicationListener Spring bean.");
}
@@ -459,7 +465,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
logger.error("Only STOMP SEND supported from within the server side. Ignoring " + message);
return;
}
sessionId = SystemStompConnectionHandler.SESSION_ID;
sessionId = SYSTEM_SESSION_ID;
stompAccessor.setSessionId(sessionId);
}
@@ -470,7 +476,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
if (StompCommand.CONNECT.equals(command)) {
if (logger.isDebugEnabled()) {
logger.debug("STOMP CONNECT in session " + sessionId + " (" + getConnectionCount() + " connections).");
logger.debug(stompAccessor.getShortLogMessage(EMPTY_PAYLOAD));
}
stompAccessor = (stompAccessor.isMutable() ? stompAccessor : StompHeaderAccessor.wrap(message));
stompAccessor.setLogin(this.clientLogin);
@@ -508,8 +514,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
@Override
public String toString() {
return "StompBrokerRelay[broker=" + this.relayHost + ":" + this.relayPort +
", " + getConnectionCount() + " connection(s)]";
return "StompBrokerRelay[" + this.relayHost + ":" + this.relayPort + "]";
}
@@ -544,8 +549,8 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
@Override
public void afterConnected(TcpConnection<byte[]> connection) {
if (logger.isInfoEnabled()) {
logger.info("TCP connection established. Forwarding: " + this.connectHeaders);
if (logger.isDebugEnabled()) {
logger.debug("TCP connection opened in session=" + getSessionId());
}
this.tcpConnection = connection;
connection.send(MessageBuilder.createMessage(EMPTY_PAYLOAD, this.connectHeaders.getMessageHeaders()));
@@ -601,24 +606,24 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
@Override
public void handleMessage(Message<byte[]> message) {
StompHeaderAccessor headerAccessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
headerAccessor.setSessionId(this.sessionId);
StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
accessor.setSessionId(this.sessionId);
if (StompCommand.CONNECTED.equals(headerAccessor.getCommand())) {
if (logger.isInfoEnabled()) {
logger.info("Received STOMP CONNECTED: " + headerAccessor);
StompCommand command = accessor.getCommand();
if (StompCommand.CONNECTED.equals(command)) {
if (logger.isDebugEnabled()) {
logger.debug("Received " + accessor.getShortLogMessage(EMPTY_PAYLOAD));
}
afterStompConnected(headerAccessor);
afterStompConnected(accessor);
}
else if (StompCommand.ERROR.equals(headerAccessor.getCommand()) && logger.isErrorEnabled()) {
logger.error("Received STOMP ERROR: " + message);
else if (logger.isErrorEnabled() && StompCommand.ERROR.equals(command)) {
logger.error("Received " + accessor.getShortLogMessage(message.getPayload()));
}
else if (logger.isTraceEnabled()) {
logger.trace(headerAccessor.isHeartbeat() ?
"Received heartbeat in session " + this.sessionId : "Received " + headerAccessor);
logger.trace("Received " + accessor.getDetailedLogMessage(message.getPayload()));
}
headerAccessor.setImmutable();
accessor.setImmutable();
sendMessageToClient(message);
}
@@ -688,8 +693,8 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
return;
}
try {
if (logger.isInfoEnabled()) {
logger.info("TCP connection to broker closed in session " + this.sessionId);
if (logger.isDebugEnabled()) {
logger.debug("TCP connection to broker closed in session " + this.sessionId);
}
sendStompErrorFrameToClient("Connection to broker closed.");
}
@@ -729,14 +734,16 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
* @return a future to wait for the result
*/
@SuppressWarnings("unchecked")
public ListenableFuture<Void> forward(Message<?> message, final StompHeaderAccessor accessor) {
public ListenableFuture<Void> forward(final Message<?> message, final StompHeaderAccessor accessor) {
TcpConnection<byte[]> conn = this.tcpConnection;
if (!this.isStompConnected) {
if (this.isRemoteClientSession) {
if (logger.isDebugEnabled()) {
logger.debug("TCP connection closed already, ignoring " + message);
logger.debug("TCP connection closed already, ignoring " +
accessor.getShortLogMessage((byte[]) message.getPayload()));
}
return EMPTY_TASK;
}
@@ -744,7 +751,8 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
throw new IllegalStateException("Cannot forward messages " +
(conn != null ? "before STOMP CONNECTED. " : "while inactive. ") +
"Consider subscribing to receive BrokerAvailabilityEvent's from " +
"an ApplicationListener Spring bean. Dropped " + message);
"an ApplicationListener Spring bean. Dropped " +
accessor.getShortLogMessage((byte[]) message.getPayload()));
}
}
@@ -752,20 +760,15 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
MessageBuilder.createMessage(message.getPayload(), accessor.getMessageHeaders()) : message;
StompCommand command = accessor.getCommand();
if (accessor.isHeartbeat()) {
logger.trace("Forwarding heartbeat in session " + this.sessionId);
}
else if (StompCommand.SUBSCRIBE.equals(command) && logger.isDebugEnabled()) {
logger.debug("Forwarding SUBSCRIBE: " + messageToSend);
}
else if (StompCommand.UNSUBSCRIBE.equals(command) && logger.isDebugEnabled()) {
logger.debug("Forwarding UNSUBSCRIBE: " + messageToSend);
}
else if (StompCommand.DISCONNECT.equals(command) && logger.isInfoEnabled()) {
logger.info("Forwarding DISCONNECT: " + messageToSend);
if (logger.isDebugEnabled() &&
StompCommand.SEND.equals(command) ||
StompCommand.SUBSCRIBE.equals(command) ||
StompCommand.UNSUBSCRIBE.equals(command) ||
StompCommand.DISCONNECT.equals(command)) {
logger.debug("Forwarding " + accessor.getShortLogMessage((byte[]) message.getPayload()));
}
else if (logger.isTraceEnabled()) {
logger.trace("Forwarding " + command + ": " + messageToSend);
logger.trace("Forwarding " + accessor.getDetailedLogMessage((byte[]) message.getPayload()));
}
ListenableFuture<Void> future = conn.send((Message<byte[]>) messageToSend);
@@ -779,10 +782,12 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
@Override
public void onFailure(Throwable t) {
if (tcpConnection != null) {
handleTcpConnectionFailure("failed to forward " + messageToSend, t);
handleTcpConnectionFailure("failed to forward " +
accessor.getShortLogMessage((byte[]) message.getPayload()), t);
}
else if (logger.isErrorEnabled()) {
logger.error("Failed to forward " + messageToSend);
logger.error("Failed to forward " +
accessor.getShortLogMessage((byte[]) message.getPayload()));
}
}
});
@@ -810,8 +815,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
*/
public void clearConnection() {
if (logger.isDebugEnabled()) {
logger.debug("Cleaning up connection state for session " + sessionId + " (" +
(getConnectionCount() - 1) + " remaining connections).");
logger.debug("Cleaning up connection state for session " + this.sessionId);
}
if (this.isRemoteClientSession) {
@@ -823,8 +827,8 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
TcpConnection<byte[]> conn = this.tcpConnection;
this.tcpConnection = null;
if (conn != null) {
if (logger.isInfoEnabled()) {
logger.info("Closing TCP connection in session " + this.sessionId);
if (logger.isDebugEnabled()) {
logger.debug("Closing TCP connection in session " + this.sessionId);
}
conn.close();
}
@@ -838,15 +842,15 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
private class SystemStompConnectionHandler extends StompConnectionHandler {
public static final String SESSION_ID = "stompRelaySystemSessionId";
public SystemStompConnectionHandler(StompHeaderAccessor connectHeaders) {
super(SESSION_ID, connectHeaders, false);
super(SYSTEM_SESSION_ID, connectHeaders, false);
}
@Override
protected void afterStompConnected(StompHeaderAccessor connectedHeaders) {
if (logger.isInfoEnabled()) {
logger.info("\"System\" session connected.");
}
super.afterStompConnected(connectedHeaders);
publishBrokerAvailableEvent();
}

22
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompCommand.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -24,6 +24,8 @@ import java.util.Map;
import org.springframework.messaging.simp.SimpMessageType;
/**
* Represents a STOMP command.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
@@ -32,15 +34,15 @@ public enum StompCommand {
// client
CONNECT,
STOMP,
SEND,
DISCONNECT,
SUBSCRIBE,
UNSUBSCRIBE,
SEND,
ACK,
NACK,
BEGIN,
COMMIT,
ABORT,
DISCONNECT,
// server
CONNECTED,
@@ -48,13 +50,8 @@ public enum StompCommand {
RECEIPT,
ERROR;
private static Map<StompCommand, SimpMessageType> messageTypes = new HashMap<StompCommand, SimpMessageType>();
private static Collection<StompCommand> destinationRequired = Arrays.asList(SEND, SUBSCRIBE, MESSAGE);
private static Collection<StompCommand> subscriptionIdRequired = Arrays.asList(SUBSCRIBE, UNSUBSCRIBE, MESSAGE);
private static Collection<StompCommand> contentLengthRequired = Arrays.asList(SEND, MESSAGE, ERROR);
private static Collection<StompCommand> bodyAllowed = Arrays.asList(SEND, MESSAGE, ERROR);
static {
messageTypes.put(StompCommand.CONNECT, SimpMessageType.CONNECT);
messageTypes.put(StompCommand.STOMP, SimpMessageType.CONNECT);
@@ -65,6 +62,13 @@ public enum StompCommand {
messageTypes.put(StompCommand.DISCONNECT, SimpMessageType.DISCONNECT);
}
private static Collection<StompCommand> destinationRequired = Arrays.asList(SEND, SUBSCRIBE, MESSAGE);
private static Collection<StompCommand> subscriptionIdRequired = Arrays.asList(SUBSCRIBE, UNSUBSCRIBE, MESSAGE);
private static Collection<StompCommand> contentLengthRequired = Arrays.asList(SEND, MESSAGE, ERROR);
private static Collection<StompCommand> bodyAllowed = Arrays.asList(SEND, MESSAGE, ERROR);
public SimpMessageType getMessageType() {
SimpMessageType type = messageTypes.get(this);
return (type != null) ? type : SimpMessageType.OTHER;

12
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompDecoder.java
View File

@@ -24,7 +24,6 @@ import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.messaging.support.MessageHeaderInitializer;
@@ -51,7 +50,8 @@ public class StompDecoder {
static final byte[] HEARTBEAT_PAYLOAD = new byte[] {'\n'};
private final Log logger = LogFactory.getLog(StompDecoder.class);
private static final Log logger = LogFactory.getLog(StompDecoder.class);
private MessageHeaderInitializer headerInitializer;
@@ -157,7 +157,7 @@ public class StompDecoder {
headerAccessor.setLeaveMutable(true);
decodedMessage = MessageBuilder.createMessage(payload, headerAccessor.getMessageHeaders());
if (logger.isTraceEnabled()) {
logger.trace("Decoded " + decodedMessage);
logger.trace("Decoded " + headerAccessor.getDetailedLogMessage(payload));
}
}
else {
@@ -176,13 +176,13 @@ public class StompDecoder {
}
}
else {
if (logger.isTraceEnabled()) {
logger.trace("Decoded heartbeat.");
}
StompHeaderAccessor headerAccessor = StompHeaderAccessor.createForHeartbeat();
initHeaders(headerAccessor);
headerAccessor.setLeaveMutable(true);
decodedMessage = MessageBuilder.createMessage(HEARTBEAT_PAYLOAD, headerAccessor.getMessageHeaders());
if (logger.isTraceEnabled()) {
logger.trace("Decoded " + headerAccessor.getDetailedLogMessage(null));
}
}
return decodedMessage;
}

2
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompEncoder.java
View File

@@ -26,7 +26,6 @@ import java.util.Map.Entry;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
@@ -74,7 +73,6 @@ public final class StompEncoder {
DataOutputStream output = new DataOutputStream(baos);
if (SimpMessageType.HEARTBEAT.equals(SimpMessageHeaderAccessor.getMessageType(headers))) {
logger.trace("Encoded heartbeat");
output.write(StompDecoder.HEARTBEAT_PAYLOAD);
}
else {

70
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompHeaderAccessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -16,6 +16,7 @@
package org.springframework.messaging.simp.stomp;
import java.nio.charset.Charset;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -62,6 +63,7 @@ public class StompHeaderAccessor extends SimpMessageHeaderAccessor {
private static final long[] DEFAULT_HEARTBEAT = new long[] {0, 0};
// STOMP header names
public static final String STOMP_ID_HEADER = "id";
@@ -438,6 +440,71 @@ public class StompHeaderAccessor extends SimpMessageHeaderAccessor {
setNativeHeader(STOMP_VERSION_HEADER, version);
}
// Logging related
@Override
public String getShortLogMessage(Object payload) {
if (StompCommand.SUBSCRIBE.equals(getCommand())) {
return "SUBSCRIBE " + getDestination() + " id=" + getSubscriptionId() + appendSession();
}
else if (StompCommand.UNSUBSCRIBE.equals(getCommand())) {
return "UNSUBSCRIBE id=" + getSubscriptionId() + appendSession();
}
else if (StompCommand.SEND.equals(getCommand())) {
return "SEND " + getDestination() + appendSession() + appendPayload(payload);
}
else if (StompCommand.CONNECT.equals(getCommand())) {
return "CONNECT" + (getUser() != null ? " user=" + getUser().getName() : "") + appendSession();
}
else if (StompCommand.CONNECTED.equals(getCommand())) {
return "CONNECTED heart-beat=" + Arrays.toString(getHeartbeat()) + appendSession();
}
else if (StompCommand.DISCONNECT.equals(getCommand())) {
return "DISCONNECT" + (getReceipt() != null ? " receipt=" + getReceipt() : "") + appendSession();
}
else {
return getDetailedLogMessage(payload);
}
}
@Override
public String getDetailedLogMessage(Object payload) {
if (isHeartbeat()) {
return "heart-beat in session " + getSessionId();
}
StompCommand command = getCommand();
if (command == null) {
return super.getDetailedLogMessage(payload);
}
StringBuilder sb = new StringBuilder();
sb.append(command.name()).append(" ").append(getNativeHeaders()).append(appendSession());
if (getUser() != null) {
sb.append(", user=").append(getUser().getName());
}
if (command.isBodyAllowed()) {
sb.append(appendPayload(payload));
}
return sb.toString();
}
private String appendSession() {
return " session=" + getSessionId();
}
private String appendPayload(Object payload) {
Assert.isInstanceOf(byte[].class, payload);
byte[] bytes = (byte[]) payload;
String contentType = (getContentType() != null ? " " + getContentType().toString() : "");
if (bytes.length == 0 || getContentType() == null || !isReadableContentType()) {
return contentType;
}
Charset charset = getContentType().getCharSet();
charset = (charset != null ? charset : StompDecoder.UTF8_CHARSET);
return (bytes.length < 80) ?
contentType + " payload=" + new String(bytes, charset) :
contentType + " payload=" + new String(Arrays.copyOf(bytes, 80), charset) + "...(truncated)";
}
private static class StompPasscode {
@@ -452,4 +519,5 @@ public class StompHeaderAccessor extends SimpMessageHeaderAccessor {
return "[PROTECTED]";
}
}
}

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

@@ -181,6 +181,10 @@ public class DefaultUserDestinationResolver implements UserDestinationResolver {
return sourceDestinationWithoutPrefix + "-user" + sessionId;
}
@Override
public String toString() {
return "DefaultUserDestinationResolver[prefix=" + this.destinationPrefix + "]";
}
private static class DestinationInfo {

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

@@ -172,7 +172,7 @@ public class UserDestinationMessageHandler implements MessageHandler, SmartLifec
Set<String> destinations = result.getTargetDestinations();
if (destinations.isEmpty()) {
if (logger.isTraceEnabled()) {
logger.trace("No user destinations for " + message);
logger.trace("No user destinations found for " + result.getSourceDestination());
}
return;
}
@@ -183,10 +183,10 @@ public class UserDestinationMessageHandler implements MessageHandler, SmartLifec
headerAccessor.setNativeHeader(header, result.getSubscribeDestination());
message = MessageBuilder.createMessage(message.getPayload(), headerAccessor.getMessageHeaders());
}
if (logger.isDebugEnabled()) {
logger.debug("Translated " + result.getSourceDestination() + " -> " + destinations);
}
for (String destination : destinations) {
if (logger.isTraceEnabled()) {
logger.trace("Sending " + message);
}
this.brokerMessagingTemplate.send(destination, message);
}
}
@@ -197,4 +197,9 @@ public class UserDestinationMessageHandler implements MessageHandler, SmartLifec
}
}
@Override
public String toString() {
return "UserDestinationMessageHandler[" + this.userDestinationResolver + "]";
}
}

5
spring-messaging/src/main/java/org/springframework/messaging/support/AbstractMessageChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -101,9 +101,6 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName
@Override
public final boolean send(Message<?> message, long timeout) {
Assert.notNull(message, "Message must not be null");
if (logger.isTraceEnabled()) {
logger.trace(this + " sending " + message);
}
message = this.interceptorChain.preSend(message, this);
if (message == null) {
return false;

4
spring-messaging/src/main/java/org/springframework/messaging/support/AbstractSubscribableChannel.java
View File

@@ -47,7 +47,7 @@ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel
boolean result = this.handlers.add(handler);
if (result) {
if (logger.isDebugEnabled()) {
logger.debug(this + " subscribed " + handler);
logger.debug(getBeanName() + " added " + handler);
}
}
return result;
@@ -58,7 +58,7 @@ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel
boolean result = this.handlers.remove(handler);
if (result) {
if (logger.isDebugEnabled()) {
logger.debug(this + " unsubscribed " + handler);
logger.debug(getBeanName() + " removed " + handler);
}
}
return result;

23
spring-messaging/src/main/java/org/springframework/messaging/support/ChannelInterceptorChain.java
View File

@@ -56,37 +56,23 @@ class ChannelInterceptorChain {
public Message<?> preSend(Message<?> message, MessageChannel channel) {
Message<?> originalMessage = message;
for (ChannelInterceptor interceptor : this.interceptors) {
message = interceptor.preSend(message, channel);
if (message == null) {
if (logger.isTraceEnabled()) {
logger.trace("preSend returned null (precluding the send)");
}
logger.debug("preSend returned null precluding send");
return null;
}
}
if (logger.isDebugEnabled()) {
if (message != originalMessage) {
logger.debug("preSend returned modified message, new message=" + message);
}
}
return message;
}
public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
if (logger.isTraceEnabled()) {
logger.trace("postSend (sent=" + sent + ")");
}
for (ChannelInterceptor interceptor : this.interceptors) {
interceptor.postSend(message, channel, sent);
}
}
public boolean preReceive(MessageChannel channel) {
if (logger.isTraceEnabled()) {
logger.trace("preReceive on channel '" + channel + "'");
}
for (ChannelInterceptor interceptor : this.interceptors) {
if (!interceptor.preReceive(channel)) {
return false;
@@ -96,12 +82,6 @@ class ChannelInterceptorChain {
}
public Message<?> postReceive(Message<?> message, MessageChannel channel) {
if (message != null && logger.isTraceEnabled()) {
logger.trace("postReceive on channel '" + channel + "', message: " + message);
}
else if (logger.isTraceEnabled()) {
logger.trace("postReceive on channel '" + channel + "', message is null");
}
for (ChannelInterceptor interceptor : this.interceptors) {
message = interceptor.postReceive(message, channel);
if (message == null) {
@@ -111,5 +91,4 @@ class ChannelInterceptorChain {
return message;
}
}

3
spring-messaging/src/main/java/org/springframework/messaging/support/GenericMessage.java
View File

@@ -94,8 +94,7 @@ public class GenericMessage<T> implements Message<T>, Serializable {
sb.append("[Payload byte[").append(((byte[]) this.payload).length).append("]]");
}
else {
sb.append("[Payload ").append(this.payload.getClass().getSimpleName());
sb.append(" content=").append(this.payload).append("]");
sb.append("[Payload=").append(this.payload).append("]");
}
sb.append("[Headers=").append(this.headers).append("]");
return sb.toString();

86
spring-messaging/src/main/java/org/springframework/messaging/support/MessageHeaderAccessor.java
View File

@@ -16,7 +16,9 @@
package org.springframework.messaging.support;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -28,9 +30,11 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.simp.stomp.StompDecoder;
import org.springframework.util.Assert;
import org.springframework.util.IdGenerator;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.PatternMatchUtils;
import org.springframework.util.StringUtils;
@@ -111,6 +115,13 @@ import org.springframework.util.StringUtils;
*/
public class MessageHeaderAccessor {
private static final MimeType[] readableMimeTypes = new MimeType[] {
MimeTypeUtils.APPLICATION_JSON, MimeTypeUtils.APPLICATION_XML,
new MimeType("text", "*"), new MimeType("application", "*+json"), new MimeType("application", "*+xml")
};
public static final Charset DEFAULT_CHARSET = Charset.forName("UTF-8");
protected Log logger = LogFactory.getLog(getClass());
private final MutableMessageHeaders headers;
@@ -469,6 +480,81 @@ public class MessageHeaderAccessor {
}
/**
* Return a concise message for logging purposes.
* @param payload the payload that corresponds to the headers.
* @return the message
*/
public String getShortLogMessage(Object payload) {
return "headers=" + this.headers.toString() + getShortPayloadLogMessage(payload);
}
/**
* Return a more detailed message for logging purposes.
* @param payload the payload that corresponds to the headers.
* @return the message
*/
public String getDetailedLogMessage(Object payload) {
return "headers=" + this.headers.toString() + getDetailedPayloadLogMessage(payload);
}
protected String getShortPayloadLogMessage(Object payload) {
if (payload instanceof String) {
String payloadText = (String) payload;
return (payloadText.length() < 80) ?
" payload=" + payloadText :
" payload=" + payloadText.substring(0, 80) + "...(truncated)";
}
else if (payload instanceof byte[]) {
byte[] bytes = (byte[]) payload;
if (isReadableContentType()) {
Charset charset = getContentType().getCharSet();
charset = (charset != null ? charset : DEFAULT_CHARSET);
return (bytes.length < 80) ?
" payload=" + new String(bytes, charset) :
" payload=" + new String(Arrays.copyOf(bytes, 80), charset) + "...(truncated)";
}
else {
return " payload=byte[" + bytes.length + "]";
}
}
else {
String payloadText = payload.toString();
return (payloadText.length() < 80) ?
" payload=" + payloadText :
" payload=" + ObjectUtils.identityToString(payload);
}
}
protected String getDetailedPayloadLogMessage(Object payload) {
if (payload instanceof String) {
return " payload=" + ((String) payload);
}
else if (payload instanceof byte[]) {
byte[] bytes = (byte[]) payload;
if (isReadableContentType()) {
Charset charset = getContentType().getCharSet();
charset = (charset != null ? charset : DEFAULT_CHARSET);
return " payload=" + new String(bytes, charset);
}
else {
return " payload=byte[" + bytes.length + "]";
}
}
else {
return " payload=" + payload;
}
}
protected boolean isReadableContentType() {
for (MimeType mimeType : readableMimeTypes) {
if (mimeType.includes(getContentType())) {
return true;
}
}
return false;
}
@Override
public String toString() {
return getClass().getSimpleName() + "[headers=" + this.headers + "]";