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

Enhance MessageHeaderAccessor support

Browse Source 
Refine semantics of ID and TIMESTAMP headers provided to protected
MessageHeaders constructor.

Refactor internal implementation of MessageHeaderAccessor.

Support mutating headers from a single thread while a message is being
built (e.g. StompDecoder creating message + then adding session id).

Improve immutablity in NativeMessageHeaderAccessor and in
StompHeaderAccessor.

Optimize object creation for initializing messages and subsequent
accessing their headers.

Introduce MessageHeaderAccessorFactory support to enable applying a
common strategies for ID and TIMESTAMP generation to every message.

Add MessageBuilder shortcut factory method for creating messages from
payload and a full-prepared MessageHeaders instance. Also add
equivalent constructors to GenericMessage and ErrorMessage.

Issue: SPR-11468
This commit is contained in:
Rossen Stoyanchev
2014-04-06 11:06:18 -04:00
parent b4995f7e4f
commit 4867546aec
29 changed files with 1825 additions and 489 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
spring-messaging/src/main/java/org/springframework/messaging/MessageHeaders.java
View File

@@ -24,7 +24,6 @@ import java.util.ArrayList;
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;
@@ -34,7 +33,6 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.util.AlternativeJdkIdGenerator;
import org.springframework.util.Assert;
import org.springframework.util.IdGenerator;
/**
@@ -70,10 +68,12 @@ import org.springframework.util.IdGenerator;
*/
public class MessageHeaders implements Map<String, Object>, Serializable {
private static final long serialVersionUID = -4615750558355702881L;
private static final long serialVersionUID = 7035068984263400920L;
private static final Log logger = LogFactory.getLog(MessageHeaders.class);
public static final UUID ID_VALUE_NONE = new UUID(0,0);
private static volatile IdGenerator idGenerator = null;
private static final IdGenerator defaultIdGenerator = new AlternativeJdkIdGenerator();
@@ -99,29 +99,43 @@ public class MessageHeaders implements Map<String, Object>, Serializable {
/**
* Consructs a {@link MessageHeaders} from the headers map; adding (or
* overwriting) the {@link #ID} and {@link #TIMESTAMP} headers.
* Construct a {@link MessageHeaders} with the given headers. An {@link #ID} and
* {@link #TIMESTAMP} headers will also be added, overriding any existing values.
*
* @param headers a map with headers to add
*/
public MessageHeaders(Map<String, Object> headers) {
this(headers, ((idGenerator != null) ? idGenerator : defaultIdGenerator).generateId(),
System.currentTimeMillis());
this(headers, null, null);
}
/**
* Constructor allowing a sub-class to access the (mutable) header map as well
* to provide the ID and TIMESTAMP header values.
* Constructor providing control over the ID and TIMESTAMP header values.
*
* @param headers a map with headers to add
* @param id the value for the {@link #ID} header, never {@code null}
* @param timestamp the value for the {@link #TIMESTAMP} header,
* or {@code null} meaning no timestamp header
* @param id the {@link #ID} header value
* @param timestamp the {@link #TIMESTAMP} header value
*/
protected MessageHeaders(Map<String, Object> headers, UUID id, Long timestamp) {
Assert.notNull(id, "'id' is required");
this.headers = (headers != null) ? new HashMap<String, Object>(headers) : new HashMap<String, Object>();
this.headers.put(ID, id);
if (timestamp != null) {
if (id == null) {
this.headers.put(ID, getIdGenerator().generateId());
}
else if (id == ID_VALUE_NONE) {
this.headers.remove(ID);
}
else {
this.headers.put(ID, id);
}
if (timestamp == null) {
this.headers.put(TIMESTAMP, System.currentTimeMillis());
}
else if (timestamp < 0) {
this.headers.remove(TIMESTAMP);
}
else {
this.headers.put(TIMESTAMP, timestamp);
}
}
@@ -131,6 +145,10 @@ public class MessageHeaders implements Map<String, Object>, Serializable {
return this.headers;
}
protected static IdGenerator getIdGenerator() {
return ((idGenerator != null) ? idGenerator : defaultIdGenerator);
}
public UUID getId() {
return this.get(ID, UUID.class);
}
@@ -179,10 +197,7 @@ public class MessageHeaders implements Map<String, Object>, Serializable {
@Override
public String 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();
return this.headers.toString();
}
/*

70
spring-messaging/src/main/java/org/springframework/messaging/simp/DefaultSimpMessageHeaderAccessorFactory.java Normal file
View File

@@ -0,0 +1,70 @@
/*
* 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.messaging.simp;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageHeaderAccessorFactorySupport;
import org.springframework.util.IdGenerator;
import java.util.UUID;
/**
* A default implementation of
* {@link org.springframework.messaging.simp.SimpMessageHeaderAccessorFactory}.
*
* @author Rossen Stoyanchev
* @since 4.1
*/
public class DefaultSimpMessageHeaderAccessorFactory extends MessageHeaderAccessorFactorySupport
implements SimpMessageHeaderAccessorFactory {
public DefaultSimpMessageHeaderAccessorFactory() {
super.setIdGenerator(ID_VALUE_NONE_GENERATOR);
}
@Override
public SimpMessageHeaderAccessor create() {
SimpMessageHeaderAccessor accessor = new SimpMessageHeaderAccessor(SimpMessageType.MESSAGE, null);
updateMessageHeaderAccessor(accessor);
return accessor;
}
@Override
public SimpMessageHeaderAccessor create(SimpMessageType messageType) {
SimpMessageHeaderAccessor accessor = new SimpMessageHeaderAccessor(messageType, null);
updateMessageHeaderAccessor(accessor);
return accessor;
}
@Override
public SimpMessageHeaderAccessor wrap(Message<?> message) {
SimpMessageHeaderAccessor accessor = new SimpMessageHeaderAccessor(message);
updateMessageHeaderAccessor(accessor);
return accessor;
}
private static final IdGenerator ID_VALUE_NONE_GENERATOR = new IdGenerator() {
@Override
public UUID generateId() {
return MessageHeaders.ID_VALUE_NONE;
}
};
}

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

@@ -21,6 +21,8 @@ import java.util.List;
import java.util.Map;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.messaging.support.NativeMessageHeaderAccessor;
import org.springframework.util.Assert;
@@ -38,6 +40,10 @@ import org.springframework.util.Assert;
*/
public class SimpMessageHeaderAccessor extends NativeMessageHeaderAccessor {
private static final SimpMessageHeaderAccessorFactory factory = new DefaultSimpMessageHeaderAccessorFactory();
// SiMP header names
public static final String CONNECT_MESSAGE_HEADER = "simpConnectMessage";
public static final String DESTINATION_HEADER = "simpDestination";
@@ -83,25 +89,32 @@ public class SimpMessageHeaderAccessor extends NativeMessageHeaderAccessor {
/**
* Create {@link SimpMessageHeaderAccessor} for a new {@link Message} with
* {@link SimpMessageType#MESSAGE}.
* Create an instance with
* {@link org.springframework.messaging.simp.SimpMessageType} {@code MESSAGE}.
*/
public static SimpMessageHeaderAccessor create() {
return new SimpMessageHeaderAccessor(SimpMessageType.MESSAGE, null);
return factory.create();
}
/**
* Create {@link SimpMessageHeaderAccessor} for a new {@link Message} of a specific type.
* Create an instance with the given
* {@link org.springframework.messaging.simp.SimpMessageType}.
*/
public static SimpMessageHeaderAccessor create(SimpMessageType messageType) {
return new SimpMessageHeaderAccessor(messageType, null);
return factory.create(messageType);
}
/**
* Create {@link SimpMessageHeaderAccessor} from the headers of an existing message.
* Create an instance from the payload and headers of the given Message.
*/
public static SimpMessageHeaderAccessor wrap(Message<?> message) {
return new SimpMessageHeaderAccessor(message);
return factory.wrap(message);
}
@Override
protected MessageHeaderAccessor createAccessor(Message<?> message) {
return factory.wrap(message);
}
public void setMessageTypeIfNotSet(SimpMessageType messageType) {
@@ -117,6 +130,13 @@ public class SimpMessageHeaderAccessor extends NativeMessageHeaderAccessor {
return (SimpMessageType) getHeader(MESSAGE_TYPE_HEADER);
}
/**
* A static alternative for access to the message type.
*/
public static SimpMessageType getMessageType(Map<String, Object> headers) {
return (SimpMessageType) headers.get(MESSAGE_TYPE_HEADER);
}
public void setDestination(String destination) {
Assert.notNull(destination, "Destination must not be null");
setHeader(DESTINATION_HEADER, destination);
@@ -129,6 +149,17 @@ public class SimpMessageHeaderAccessor extends NativeMessageHeaderAccessor {
return (String) getHeader(DESTINATION_HEADER);
}
/**
* A static alternative for access to the destination header.
*/
public static String getDestination(Map<String, Object> headers) {
return (String) headers.get(DESTINATION_HEADER);
}
public void setSubscriptionId(String subscriptionId) {
setHeader(SUBSCRIPTION_ID_HEADER, subscriptionId);
}
/**
* @return the subscription id (if any) of the message
*/
@@ -136,8 +167,15 @@ public class SimpMessageHeaderAccessor extends NativeMessageHeaderAccessor {
return (String) getHeader(SUBSCRIPTION_ID_HEADER);
}
public void setSubscriptionId(String subscriptionId) {
setHeader(SUBSCRIPTION_ID_HEADER, subscriptionId);
/**
* A static alternative for access to the subscription id header.
*/
public static String getSubscriptionId(Map<String, Object> headers) {
return (String) headers.get(SUBSCRIPTION_ID_HEADER);
}
public void setSessionId(String sessionId) {
setHeader(SESSION_ID_HEADER, sessionId);
}
/**
@@ -147,8 +185,18 @@ public class SimpMessageHeaderAccessor extends NativeMessageHeaderAccessor {
return (String) getHeader(SESSION_ID_HEADER);
}
public void setSessionId(String sessionId) {
setHeader(SESSION_ID_HEADER, sessionId);
/**
* A static alternative for access to the session id header.
*/
public static String getSessionId(Map<String, Object> headers) {
return (String) headers.get(SESSION_ID_HEADER);
}
/**
* A static alternative for access to the session attributes header.
*/
public void setSessionAttributes(Map<String, Object> attributes) {
setHeader(SESSION_ATTRIBUTES, attributes);
}
/**
@@ -159,8 +207,16 @@ public class SimpMessageHeaderAccessor extends NativeMessageHeaderAccessor {
return (Map<String, Object>) getHeader(SESSION_ATTRIBUTES);
}
public void setSessionAttributes(Map<String, Object> attributes) {
setHeader(SESSION_ATTRIBUTES, attributes);
/**
* A static alternative for access to the session attributes header.
*/
@SuppressWarnings("unchecked")
public static Map<String, Object> getSessionAttributes(Map<String, Object> headers) {
return (Map<String, Object>) headers.get(SESSION_ATTRIBUTES);
}
public void setUser(Principal principal) {
setHeader(USER_HEADER, principal);
}
/**
@@ -170,8 +226,11 @@ public class SimpMessageHeaderAccessor extends NativeMessageHeaderAccessor {
return (Principal) getHeader(USER_HEADER);
}
public void setUser(Principal principal) {
setHeader(USER_HEADER, principal);
/**
* A static alternative for access to the user header.
*/
public static Principal getUser(Map<String, Object> headers) {
return (Principal) headers.get(USER_HEADER);
}
}

48
spring-messaging/src/main/java/org/springframework/messaging/simp/SimpMessageHeaderAccessorFactory.java Normal file
View File

@@ -0,0 +1,48 @@
/*
* 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.messaging.simp;
import org.springframework.messaging.Message;
/**
* A factory for creating pre-configured instances of type
* {@link org.springframework.messaging.simp.SimpMessageHeaderAccessor}.
*
* @author Rossen Stoyanchev
* @since 4.1
*/
public interface SimpMessageHeaderAccessorFactory {
/**
* Create an instance with
* {@link org.springframework.messaging.simp.SimpMessageType} {@code MESSAGE}.
*/
SimpMessageHeaderAccessor create();
/**
* Create an instance with the given
* {@link org.springframework.messaging.simp.SimpMessageType}.
*/
SimpMessageHeaderAccessor create(SimpMessageType messageType);
/**
* Create an instance from the payload and headers of the given Message.
*/
SimpMessageHeaderAccessor wrap(Message<?> message);
}

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

@@ -164,7 +164,7 @@ public abstract class AbstractBrokerMessageHandler
public final void handleMessage(Message<?> message) {
if (!this.running) {
if (logger.isTraceEnabled()) {
logger.trace("Message broker is not running. Ignoring message id=" + message.getHeaders().getId());
logger.trace("Message broker is not running. Ignoring message=" + message);
}
return;
}

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

@@ -123,7 +123,7 @@ public class BufferingStompDecoder extends StompDecoder {
if (bufferToDecode.hasRemaining()) {
this.chunks.add(bufferToDecode);
this.expectedContentLength = getContentLength(headers);
this.expectedContentLength = StompHeaderAccessor.getContentLength(headers);
}
return messages;

78
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/DefaultStompHeaderAccessorFactory.java Normal file
View File

@@ -0,0 +1,78 @@
/*
* 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.messaging.simp.stomp;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageHeaderAccessorFactorySupport;
import org.springframework.util.IdGenerator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
/**
* A default implementation of
* {@link org.springframework.messaging.simp.stomp.StompHeaderAccessorFactory}.
*
* @author Rossen Stoyanchev
* @since 4.1
*/
public class DefaultStompHeaderAccessorFactory extends MessageHeaderAccessorFactorySupport
implements StompHeaderAccessorFactory {
public DefaultStompHeaderAccessorFactory() {
super.setIdGenerator(ID_VALUE_NONE_GENERATOR);
}
@Override
public StompHeaderAccessor create(StompCommand command) {
StompHeaderAccessor accessor = new StompHeaderAccessor(command, null);
updateMessageHeaderAccessor(accessor);
return accessor;
}
@Override
public StompHeaderAccessor create(StompCommand command, Map<String, List<String>> headers) {
StompHeaderAccessor accessor = new StompHeaderAccessor(command, headers);
updateMessageHeaderAccessor(accessor);
return accessor;
}
@Override
public StompHeaderAccessor createForHeartbeat() {
StompHeaderAccessor accessor = new StompHeaderAccessor();
updateMessageHeaderAccessor(accessor);
return accessor;
}
@Override
public StompHeaderAccessor wrap(Message<?> message) {
StompHeaderAccessor accessor = new StompHeaderAccessor(message);
updateMessageHeaderAccessor(accessor);
return accessor;
}
private static final IdGenerator ID_VALUE_NONE_GENERATOR = new IdGenerator() {
@Override
public UUID generateId() {
return MessageHeaders.ID_VALUE_NONE;
}
};
}

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

@@ -26,10 +26,9 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.messaging.support.NativeMessageHeaderAccessor;
import org.springframework.util.InvalidMimeTypeException;
import org.springframework.util.MultiValueMap;
/**
@@ -47,9 +46,9 @@ import org.springframework.util.MultiValueMap;
*/
public class StompDecoder {
private static final Charset UTF8_CHARSET = Charset.forName("UTF-8");
static final Charset UTF8_CHARSET = Charset.forName("UTF-8");
private static final byte[] HEARTBEAT_PAYLOAD = new byte[] {'\n'};
static final byte[] HEARTBEAT_PAYLOAD = new byte[] {'\n'};
private final Log logger = LogFactory.getLog(StompDecoder.class);
@@ -65,7 +64,7 @@ public class StompDecoder {
* @return the decoded messages or an empty list
*/
public List<Message<byte[]>> decode(ByteBuffer buffer) {
return decode(buffer, new LinkedMultiValueMap<String, String>());
return decode(buffer, null);
}
/**
@@ -79,27 +78,25 @@ public class StompDecoder {
* <p>If the buffer contains one ore more STOMP frames, those are returned and
* the buffer reset to point to the beginning of the unused partial content.
*
* <p>The input headers map is used to store successfully parsed headers and
* is cleared after ever successfully read message. So when partial content is
* read the caller can check if a "content-length" header was read, which helps
* to determine how much more content is needed before the next STOMP frame
* can be decoded.
* <p>The output partialMessageHeaders map is used to store successfully parsed
* headers in case of partial content. The caller can then check if a
* "content-length" header was read, which helps to determine how much more
* content is needed before the next attempt to decode.
*
* @param buffer The buffer to decode the STOMP frame from
* @param headers an empty map that will contain successfully parsed headers
* @param partialMessageHeaders an empty output map that will store the last
* successfully parsed partialMessageHeaders in case of partial message content
* in cases where the partial buffer ended with a partial STOMP frame
*
* @return decoded messages or an empty list
* @throws StompConversionException raised in case of decoding issues
*/
public List<Message<byte[]>> decode(ByteBuffer buffer, MultiValueMap<String, String> headers) {
Assert.notNull(headers, "headers is required");
public List<Message<byte[]>> decode(ByteBuffer buffer, MultiValueMap<String, String> partialMessageHeaders) {
List<Message<byte[]>> messages = new ArrayList<Message<byte[]>>();
while (buffer.hasRemaining()) {
Message<byte[]> m = decodeMessage(buffer, headers);
Message<byte[]> m = decodeMessage(buffer, partialMessageHeaders);
if (m != null) {
messages.add(m);
headers.clear();
}
else {
break;
@@ -120,17 +117,25 @@ public class StompDecoder {
String command = readCommand(buffer);
if (command.length() > 0) {
readHeaders(buffer, headers);
byte[] payload = readPayload(buffer, headers);
StompHeaderAccessor headerAccessor = null;
byte[] payload = null;
if (buffer.remaining() > 0) {
StompCommand stompCommand = StompCommand.valueOf(command);
headerAccessor = StompHeaderAccessor.create(stompCommand);
readHeaders(buffer, headerAccessor);
payload = readPayload(buffer, headerAccessor);
}
if (payload != null) {
StompCommand stompCommand = StompCommand.valueOf(command);
if ((payload.length > 0) && (!stompCommand.isBodyAllowed())) {
throw new StompConversionException(stompCommand + " shouldn't have but " +
"has a payload with length=" + payload.length + ", headers=" + headers);
if ((payload.length > 0) && (!headerAccessor.getCommand().isBodyAllowed())) {
throw new StompConversionException(headerAccessor.getCommand() +
" shouldn't have a payload: length=" + payload.length + ", headers=" + headers);
}
decodedMessage = MessageBuilder.withPayload(payload)
.setHeaders(StompHeaderAccessor.create(stompCommand, headers)).build();
headerAccessor.updateSimpMessageHeadersFromStompHeaders();
headerAccessor.setLeaveMutable(true);
decodedMessage = MessageBuilder.createMessage(payload, headerAccessor.getMessageHeaders());
if (logger.isDebugEnabled()) {
logger.debug("Decoded " + decodedMessage);
}
@@ -139,6 +144,14 @@ public class StompDecoder {
if (logger.isTraceEnabled()) {
logger.trace("Received incomplete frame. Resetting buffer.");
}
if (headers != null && headerAccessor != null) {
String name = NativeMessageHeaderAccessor.NATIVE_HEADERS;
@SuppressWarnings("unchecked")
MultiValueMap<String, String> map = (MultiValueMap<String, String>) headerAccessor.getHeader(name);
if (map != null) {
headers.putAll(map);
}
}
buffer.reset();
}
}
@@ -146,8 +159,9 @@ public class StompDecoder {
if (logger.isTraceEnabled()) {
logger.trace("Decoded heartbeat");
}
decodedMessage = MessageBuilder.withPayload(HEARTBEAT_PAYLOAD).setHeaders(
StompHeaderAccessor.create(SimpMessageType.HEARTBEAT)).build();
StompHeaderAccessor headerAccessor = StompHeaderAccessor.createForHeartbeat();
headerAccessor.setLeaveMutable(true);
decodedMessage = MessageBuilder.createMessage(HEARTBEAT_PAYLOAD, headerAccessor.getMessageHeaders());
}
return decodedMessage;
}
@@ -173,7 +187,7 @@ public class StompDecoder {
return new String(command.toByteArray(), UTF8_CHARSET);
}
private void readHeaders(ByteBuffer buffer, MultiValueMap<String, String> headers) {
private void readHeaders(ByteBuffer buffer, StompHeaderAccessor headerAccessor) {
while (true) {
ByteArrayOutputStream headerStream = new ByteArrayOutputStream(256);
while (buffer.remaining() > 0 && !tryConsumeEndOfLine(buffer)) {
@@ -191,7 +205,14 @@ public class StompDecoder {
else {
String headerName = unescape(header.substring(0, colonIndex));
String headerValue = unescape(header.substring(colonIndex + 1));
headers.add(headerName, headerValue);
try {
headerAccessor.addNativeHeader(headerName, headerValue);
}
catch (InvalidMimeTypeException ex) {
if (buffer.remaining() > 0) {
throw ex;
}
}
}
}
else {
@@ -237,8 +258,17 @@ public class StompDecoder {
return sb.toString();
}
private byte[] readPayload(ByteBuffer buffer, MultiValueMap<String, String> headers) {
Integer contentLength = getContentLength(headers);
private byte[] readPayload(ByteBuffer buffer, StompHeaderAccessor headerAccessor) {
Integer contentLength;
try {
contentLength = headerAccessor.getContentLength();
}
catch (NumberFormatException ex) {
logger.warn("Ignoring invalid content-length: '" + headerAccessor);
contentLength = null;
}
if (contentLength != null && contentLength >= 0) {
if (buffer.remaining() > contentLength) {
byte[] payload = new byte[contentLength];
@@ -267,19 +297,6 @@ public class StompDecoder {
return null;
}
protected Integer getContentLength(MultiValueMap<String, String> headers) {
if (headers.containsKey(StompHeaderAccessor.STOMP_CONTENT_LENGTH_HEADER)) {
String rawContentLength = headers.getFirst(StompHeaderAccessor.STOMP_CONTENT_LENGTH_HEADER);
try {
return Integer.valueOf(rawContentLength);
}
catch (NumberFormatException ex) {
logger.warn("Ignoring invalid content-length header value: '" + rawContentLength + "'");
}
}
return null;
}
/**
* Try to read an EOL incrementing the buffer position if successful.
*

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

@@ -19,7 +19,7 @@ package org.springframework.messaging.simp.stomp;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -28,7 +28,10 @@ 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;
import org.springframework.messaging.support.NativeMessageHeaderAccessor;
import org.springframework.util.Assert;
/**
* An encoder for STOMP frames.
@@ -43,32 +46,45 @@ public final class StompEncoder {
private static final byte COLON = ':';
private static final Charset UTF8_CHARSET = Charset.forName("UTF-8");
private final Log logger = LogFactory.getLog(StompEncoder.class);
/**
* Encodes the given STOMP {@code message} into a {@code byte[]}
*
* @param message the message to encode
* @return the encoded message
*/
public byte[] encode(Message<byte[]> message) {
return encode(message.getHeaders(), message.getPayload());
}
/**
* Encodes the given payload and headers into a {@code byte[]}.
*
* @param headers the headers
* @param payload the payload
* @return the encoded message
*/
public byte[] encode(Map<String, Object> headers, byte[] payload) {
Assert.notNull(headers, "'headers' is required");
Assert.notNull(payload, "'payload' is required");
try {
ByteArrayOutputStream baos = new ByteArrayOutputStream(128 + message.getPayload().length);
ByteArrayOutputStream baos = new ByteArrayOutputStream(128 + payload.length);
DataOutputStream output = new DataOutputStream(baos);
StompHeaderAccessor headers = StompHeaderAccessor.wrap(message);
if (SimpMessageType.HEARTBEAT == headers.getMessageType()) {
if (SimpMessageType.HEARTBEAT.equals(SimpMessageHeaderAccessor.getMessageType(headers))) {
logger.trace("Encoded heartbeat");
output.write(message.getPayload());
output.write(StompDecoder.HEARTBEAT_PAYLOAD);
}
else {
output.write(headers.getCommand().toString().getBytes(UTF8_CHARSET));
StompCommand command = StompHeaderAccessor.getCommand(headers);
Assert.notNull(command, "Missing STOMP command: " + headers);
output.write(command.toString().getBytes(StompDecoder.UTF8_CHARSET));
output.write(LF);
writeHeaders(headers, message, output);
writeHeaders(command, headers, payload, output);
output.write(LF);
writeBody(message, output);
writeBody(payload, output);
output.write((byte) 0);
}
@@ -79,20 +95,30 @@ public final class StompEncoder {
}
}
private void writeHeaders(StompHeaderAccessor headers, Message<byte[]> message, DataOutputStream output)
private void writeHeaders(StompCommand command, Map<String, Object> headers, byte[] payload, DataOutputStream output)
throws IOException {
StompCommand command = headers.getCommand();
Map<String,List<String>> stompHeaders = headers.toStompHeaderMap();
boolean shouldEscape = (command != StompCommand.CONNECT && command != StompCommand.CONNECTED);
@SuppressWarnings("unchecked")
Map<String,List<String>> nativeHeaders =
(Map<String, List<String>>) headers.get(NativeMessageHeaderAccessor.NATIVE_HEADERS);
if (logger.isDebugEnabled()) {
logger.debug("Encoded STOMP " + command + ", headers=" + stompHeaders);
logger.debug("Encoding STOMP " + command + ", headers=" + nativeHeaders);
}
for (Entry<String, List<String>> entry : stompHeaders.entrySet()) {
if (nativeHeaders == null) {
return;
}
boolean shouldEscape = (command != StompCommand.CONNECT && command != StompCommand.CONNECTED);
for (Entry<String, List<String>> entry : nativeHeaders.entrySet()) {
byte[] key = encodeHeaderString(entry.getKey(), shouldEscape);
for (String value : entry.getValue()) {
List<String> values = entry.getValue();
if (StompHeaderAccessor.STOMP_PASSCODE_HEADER.equals(entry.getKey())) {
values = Arrays.asList(StompHeaderAccessor.getPasscode(headers));
}
for (String value : values) {
output.write(key);
output.write(COLON);
output.write(encodeHeaderString(value, shouldEscape));
@@ -100,16 +126,16 @@ public final class StompEncoder {
}
}
if (command.requiresContentLength()) {
int contentLength = message.getPayload().length;
output.write("content-length:".getBytes(UTF8_CHARSET));
output.write(Integer.toString(contentLength).getBytes(UTF8_CHARSET));
int contentLength = payload.length;
output.write("content-length:".getBytes(StompDecoder.UTF8_CHARSET));
output.write(Integer.toString(contentLength).getBytes(StompDecoder.UTF8_CHARSET));
output.write(LF);
}
}
private byte[] encodeHeaderString(String input, boolean escape) {
input = escape ? escape(input) : input;
return input.getBytes(UTF8_CHARSET);
return input.getBytes(StompDecoder.UTF8_CHARSET);
}
/**
@@ -139,8 +165,8 @@ public final class StompEncoder {
return sb.toString();
}
private void writeBody(Message<byte[]> message, DataOutputStream output) throws IOException {
output.write(message.getPayload());
private void writeBody(byte[] payload, DataOutputStream output) throws IOException {
output.write(payload);
}
}

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

@@ -26,27 +26,45 @@ import java.util.concurrent.atomic.AtomicLong;
import org.springframework.messaging.Message;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import org.springframework.util.MultiValueMap;
import org.springframework.util.StringUtils;
/**
* Can be used to prepare headers for a new STOMP message, or to access and/or modify
* STOMP-specific headers of an existing message.
* A {@code MessageHeaderAccessor} to use when creating a {@code Message} from a
* decoded STOMP frame, or when encoding a {@code Message} to a STOMP frame.
*
* <p>Use one of the static factory method in this class, then call getters and setters,
* and at the end if necessary call {@link #toMap()} to obtain the updated headers
* or call {@link #toNativeHeaderMap()} to obtain only the STOMP-specific headers.
* <p>When created from STOMP frame content, the actual STOMP headers are stored
* in the native header sub-map managed by the parent class
* {@link org.springframework.messaging.support.NativeMessageHeaderAccessor}
* while the parent class
* {@link org.springframework.messaging.simp.SimpMessageHeaderAccessor} manages
* common processing headers some of which are based on STOMP headers (e.g.
* destination, content-type, etc).
*
* <p>An instance of this class can also be created by wrapping an existing
* {@code Message}. That message may have been created with the more generic
* {@link org.springframework.messaging.simp.SimpMessageHeaderAccessor} in
* which case STOMP headers are created from common processing headers.
* In this case it is also necessary to invoke either
* {@link #updateStompCommandAsClientMessage()} or
* {@link #updateStompCommandAsServerMessage()} if sending a message and
* depending on whether a message is sent to a client or the message broker.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
public class StompHeaderAccessor extends SimpMessageHeaderAccessor {
private static final StompHeaderAccessorFactory factory = new DefaultStompHeaderAccessorFactory();
private static final AtomicLong messageIdCounter = new AtomicLong();
private static final long[] DEFAULT_HEARTBEAT = new long[] {0, 0};
// STOMP header names
public static final String STOMP_ID_HEADER = "id";
@@ -83,9 +101,6 @@ public class StompHeaderAccessor extends SimpMessageHeaderAccessor {
public static final String STOMP_HEARTBEAT_HEADER = "heart-beat";
private static final long[] DEFAULT_HEARTBEAT = new long[] {0, 0};
// Other header names
private static final String COMMAND_HEADER = "stompCommand";
@@ -94,135 +109,113 @@ public class StompHeaderAccessor extends SimpMessageHeaderAccessor {
/**
* A constructor for creating new STOMP message headers.
* A constructor for creating message headers from a parsed STOMP frame.
*/
private StompHeaderAccessor(StompCommand command, Map<String, List<String>> externalSourceHeaders) {
StompHeaderAccessor(StompCommand command, Map<String, List<String>> externalSourceHeaders) {
super(command.getMessageType(), externalSourceHeaders);
Assert.notNull(command, "Command must not be null");
setHeader(COMMAND_HEADER, command);
if (externalSourceHeaders != null) {
setSimpMessageHeaders(command, externalSourceHeaders);
}
}
private void setSimpMessageHeaders(StompCommand command, Map<String, List<String>> extHeaders) {
List<String> values = extHeaders.get(StompHeaderAccessor.STOMP_DESTINATION_HEADER);
if (!CollectionUtils.isEmpty(values)) {
super.setDestination(values.get(0));
}
values = extHeaders.get(StompHeaderAccessor.STOMP_CONTENT_TYPE_HEADER);
if (!CollectionUtils.isEmpty(values)) {
super.setContentType(MimeTypeUtils.parseMimeType(values.get(0)));
}
if (StompCommand.SUBSCRIBE.equals(command) || StompCommand.UNSUBSCRIBE.equals(command)) {
values = extHeaders.get(StompHeaderAccessor.STOMP_ID_HEADER);
if (!CollectionUtils.isEmpty(values)) {
super.setSubscriptionId(values.get(0));
}
}
else if (StompCommand.MESSAGE.equals(command)) {
values = extHeaders.get(StompHeaderAccessor.STOMP_SUBSCRIPTION_HEADER);
if (!CollectionUtils.isEmpty(values)) {
super.setSubscriptionId(values.get(0));
}
}
else if (StompCommand.CONNECT.equals(command)) {
if (!StringUtils.isEmpty(getPasscode())) {
setHeader(CREDENTIALS_HEADER, new StompPasscode(getPasscode()));
setPasscode("PROTECTED");
}
}
updateSimpMessageHeadersFromStompHeaders();
}
/**
* A constructor for accessing and modifying existing message headers.
* Note that the message headers may not have been created from a STOMP frame
* but may have rather originated from using the more generic
* {@link org.springframework.messaging.simp.SimpMessageHeaderAccessor}.
*/
private StompHeaderAccessor(Message<?> message) {
StompHeaderAccessor(Message<?> message) {
super(message);
updateStompHeadersFromSimpMessageHeaders();
}
StompHeaderAccessor() {
super(SimpMessageType.HEARTBEAT, null);
}
void updateSimpMessageHeadersFromStompHeaders() {
if (getNativeHeaders() == null) {
return;
}
String value = getFirstNativeHeader(STOMP_DESTINATION_HEADER);
if (value != null) {
super.setDestination(value);
}
value = getFirstNativeHeader(STOMP_CONTENT_TYPE_HEADER);
if (value != null) {
super.setContentType(MimeTypeUtils.parseMimeType(value));
}
StompCommand command = getCommand();
if (StompCommand.MESSAGE.equals(command)) {
value = getFirstNativeHeader(STOMP_SUBSCRIPTION_HEADER);
if (value != null) {
super.setSubscriptionId(value);
}
}
else if (StompCommand.SUBSCRIBE.equals(command) || StompCommand.UNSUBSCRIBE.equals(command)) {
value = getFirstNativeHeader(STOMP_ID_HEADER);
if (value != null) {
super.setSubscriptionId(value);
}
}
else if (StompCommand.CONNECT.equals(command)) {
protectPasscode();
}
}
private void updateStompHeadersFromSimpMessageHeaders() {
if (getDestination() != null) {
setNativeHeader(STOMP_DESTINATION_HEADER, getDestination());
}
if (getContentType() != null) {
setNativeHeader(STOMP_CONTENT_TYPE_HEADER, getContentType().toString());
}
trySetStompHeaderForSubscriptionId();
}
/**
* Create {@link StompHeaderAccessor} for a new {@link Message}.
* Create an instance for the given STOMP command.
*/
public static StompHeaderAccessor create(StompCommand command) {
return new StompHeaderAccessor(command, null);
return factory.create(command);
}
/**
* Create {@link StompHeaderAccessor} from parsed STOP frame content.
* Create an instance for the given STOMP command and headers.
*/
public static StompHeaderAccessor create(StompCommand command, Map<String, List<String>> headers) {
return new StompHeaderAccessor(command, headers);
return factory.create(command, headers);
}
/**
* Create {@link StompHeaderAccessor} from the headers of an existing {@link Message}.
* Create headers for a heartbeat. While a STOMP heartbeat frame does not
* have headers, a session id is needed for processing purposes at a minimum.
*/
public static StompHeaderAccessor createForHeartbeat() {
return factory.createForHeartbeat();
}
/**
* Create an instance from the payload and headers of the given Message.
*/
public static StompHeaderAccessor wrap(Message<?> message) {
return new StompHeaderAccessor(message);
return factory.wrap(message);
}
/**
* Return STOMP headers including original, wrapped STOMP headers (if any) plus
* additional header updates made through accessor methods.
*/
@Override
public Map<String, List<String>> toNativeHeaderMap() {
Map<String, List<String>> result = super.toNativeHeaderMap();
String destination = super.getDestination();
if (destination != null) {
result.put(STOMP_DESTINATION_HEADER, Arrays.asList(destination));
}
MimeType contentType = super.getContentType();
if (contentType != null) {
result.put(STOMP_CONTENT_TYPE_HEADER, Arrays.asList(contentType.toString()));
}
if (getCommand() != null && getCommand().requiresSubscriptionId()) {
String subscriptionId = getSubscriptionId();
if (subscriptionId != null) {
String name = StompCommand.MESSAGE.equals(getCommand()) ? STOMP_SUBSCRIPTION_HEADER : STOMP_ID_HEADER;
result.put(name, Arrays.asList(subscriptionId));
}
else {
logger.warn(getCommand() + " frame does not have a subscription identifier" + this.toString());
}
}
if (StompCommand.MESSAGE.equals(getCommand()) && ((getMessageId() == null))) {
String messageId = getSessionId() + "-" + messageIdCounter.getAndIncrement();
result.put(STOMP_MESSAGE_ID_HEADER, Arrays.asList(messageId));
}
return result;
protected MessageHeaderAccessor createAccessor(Message<?> message) {
return factory.wrap(message);
}
public Map<String, List<String>> toStompHeaderMap() {
if (StompCommand.CONNECT.equals(getCommand())) {
StompPasscode credentials = (StompPasscode) getHeader(CREDENTIALS_HEADER);
if (credentials != null) {
Map<String, List<String>> headers = toNativeHeaderMap();
headers.put(STOMP_PASSCODE_HEADER, Arrays.asList(credentials.passcode));
return headers;
}
}
return toNativeHeaderMap();
Map<String, List<String>> getNativeHeaders() {
@SuppressWarnings("unchecked")
Map<String, List<String>> map = (Map<String, List<String>>) getHeader(NATIVE_HEADERS);
return (map != null ? map : Collections.<String, List<String>>emptyMap());
}
public StompCommand updateStompCommandAsClientMessage() {
Assert.state(SimpMessageType.MESSAGE.equals(getMessageType()),
"Unexpected message type " + getMessage());
Assert.state(SimpMessageType.MESSAGE.equals(getMessageType()), "Unexpected message type " + getMessage());
if (getCommand() == null) {
setHeader(COMMAND_HEADER, StompCommand.SEND);
@@ -236,26 +229,47 @@ public class StompHeaderAccessor extends SimpMessageHeaderAccessor {
public void updateStompCommandAsServerMessage() {
Assert.state(SimpMessageType.MESSAGE.equals(getMessageType()),
"Unexpected message type " + getMessage());
Assert.state(SimpMessageType.MESSAGE.equals(getMessageType()), "Unexpected message type " + getMessage());
if ((getCommand() == null) || getCommand().equals(StompCommand.SEND)) {
StompCommand command = getCommand();
if ((command == null) || StompCommand.SEND.equals(command)) {
setHeader(COMMAND_HEADER, StompCommand.MESSAGE);
}
else if (!getCommand().equals(StompCommand.MESSAGE)) {
throw new IllegalStateException("Unexpected STOMP command " + getCommand());
else if (!StompCommand.MESSAGE.equals(command)) {
throw new IllegalStateException("Unexpected STOMP command " + command);
}
trySetStompHeaderForSubscriptionId();
if (getMessageId() == null) {
String messageId = getSessionId() + "-" + messageIdCounter.getAndIncrement();
setNativeHeader(STOMP_MESSAGE_ID_HEADER, messageId);
}
}
/**
* @return the STOMP command, or {@code null} if not yet set.
*/
public StompCommand getCommand() {
return (StompCommand) getHeader(COMMAND_HEADER);
}
/**
* A static alternative for access to the STOMP command.
*/
public static StompCommand getCommand(Map<String, Object> headers) {
return (StompCommand) headers.get(COMMAND_HEADER);
}
public Set<String> getAcceptVersion() {
String rawValue = getFirstNativeHeader(STOMP_ACCEPT_VERSION_HEADER);
return (rawValue != null) ? StringUtils.commaDelimitedListToSet(rawValue) : Collections.<String>emptySet();
}
public boolean isHeartbeat() {
return (SimpMessageType.HEARTBEAT == getMessageType());
}
public void setAcceptVersion(String acceptVersion) {
setNativeHeader(STOMP_ACCEPT_VERSION_HEADER, acceptVersion);
}
@@ -288,9 +302,41 @@ public class StompHeaderAccessor extends SimpMessageHeaderAccessor {
setNativeHeader(STOMP_CONTENT_TYPE_HEADER, contentType.toString());
}
@Override
public void setSubscriptionId(String subscriptionId) {
super.setSubscriptionId(subscriptionId);
trySetStompHeaderForSubscriptionId();
}
private void trySetStompHeaderForSubscriptionId() {
String subscriptionId = getSubscriptionId();
if (subscriptionId != null) {
if (getCommand() != null && StompCommand.MESSAGE.equals(getCommand())) {
setNativeHeader(STOMP_SUBSCRIPTION_HEADER, subscriptionId);
}
else {
SimpMessageType messageType = getMessageType();
if (SimpMessageType.SUBSCRIBE.equals(messageType) || SimpMessageType.UNSUBSCRIBE.equals(messageType)) {
setNativeHeader(STOMP_ID_HEADER, subscriptionId);
}
}
}
}
public Integer getContentLength() {
String contentLength = getFirstNativeHeader(STOMP_CONTENT_LENGTH_HEADER);
return StringUtils.hasText(contentLength) ? new Integer(contentLength) : null;
if (containsNativeHeader(STOMP_CONTENT_LENGTH_HEADER)) {
return Integer.valueOf(getFirstNativeHeader(STOMP_CONTENT_LENGTH_HEADER));
}
return null;
}
public static Integer getContentLength(Map<String, List<String>> nativeHeaders) {
if (nativeHeaders.containsKey(STOMP_CONTENT_LENGTH_HEADER)) {
List<String> values = nativeHeaders.get(STOMP_CONTENT_LENGTH_HEADER);
String value = (values != null ? values.get(0) : null);
return Integer.valueOf(value);
}
return null;
}
public void setContentLength(int contentLength) {
@@ -298,7 +344,7 @@ public class StompHeaderAccessor extends SimpMessageHeaderAccessor {
}
public void setHeartbeat(long cx, long cy) {
setNativeHeader(STOMP_HEARTBEAT_HEADER, StringUtils.arrayToCommaDelimitedString(new Object[] {cx, cy}));
setNativeHeader(STOMP_HEARTBEAT_HEADER, StringUtils.arrayToCommaDelimitedString(new Object[]{cx, cy}));
}
public void setAck(String ack) {
@@ -328,10 +374,31 @@ public class StompHeaderAccessor extends SimpMessageHeaderAccessor {
public void setPasscode(String passcode) {
setNativeHeader(STOMP_PASSCODE_HEADER, passcode);
protectPasscode();
}
private void protectPasscode() {
String value = getFirstNativeHeader(STOMP_PASSCODE_HEADER);
if (value != null && !"PROTECTED".equals(value)) {
setHeader(CREDENTIALS_HEADER, new StompPasscode(value));
setNativeHeader(STOMP_PASSCODE_HEADER, "PROTECTED");
}
}
/**
* @return the passcode header value or {@code null}.
*/
public String getPasscode() {
return getFirstNativeHeader(STOMP_PASSCODE_HEADER);
StompPasscode credentials = (StompPasscode) getHeader(CREDENTIALS_HEADER);
return (credentials != null ? credentials.passcode : null);
}
/**
* A static alternative for access to the passcode header.
*/
public static String getPasscode(Map<String, Object> headers) {
StompPasscode credentials = (StompPasscode) headers.get(CREDENTIALS_HEADER);
return (credentials != null ? credentials.passcode : null);
}
public void setReceiptId(String receiptId) {

55
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompHeaderAccessorFactory.java Normal file
View File

@@ -0,0 +1,55 @@
/*
* 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.messaging.simp.stomp;
import org.springframework.messaging.Message;
import java.util.List;
import java.util.Map;
/**
* A factory for creating pre-configured instances of type
* {@link org.springframework.messaging.simp.stomp.StompHeaderAccessor}.
*
* @author Rossen Stoyanchev
* @since 4.1
*/
public interface StompHeaderAccessorFactory {
/**
* Create an instance for the given STOMP command.
*/
StompHeaderAccessor create(StompCommand command);
/**
* Create an instance for the given STOMP command and headers.
*/
StompHeaderAccessor create(StompCommand command, Map<String, List<String>> headers);
/**
* Create headers for a heartbeat. While a STOMP heartbeat frame does not
* have headers, a session id is needed for processing purposes at a minimum.
*/
StompHeaderAccessor createForHeartbeat();
/**
* Create an instance from the payload and headers of the given Message.
*/
StompHeaderAccessor wrap(Message<?> message);
}

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

@@ -104,7 +104,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 id=" + message.getHeaders().getId());
logger.trace("[" + this.beanName + "] sending message=" + message);
}
message = this.interceptorChain.preSend(message, this);

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

@@ -57,7 +57,7 @@ class ChannelInterceptorChain {
public Message<?> preSend(Message<?> message, MessageChannel channel) {
UUID originalId = message.getHeaders().getId();
Message<?> originalMessage = message;
for (ChannelInterceptor interceptor : this.interceptors) {
message = interceptor.preSend(message, channel);
if (message == null) {
@@ -68,8 +68,8 @@ class ChannelInterceptorChain {
}
}
if (logger.isDebugEnabled()) {
if (!message.getHeaders().getId().equals(originalId)) {
logger.debug("preSend returned modified message, new message id=" + message.getHeaders().getId());
if (message != originalMessage) {
logger.debug("preSend returned modified message, new message=" + message);
}
}
return message;
@@ -77,7 +77,7 @@ class ChannelInterceptorChain {
public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
if (logger.isTraceEnabled()) {
logger.trace("postSend (sent=" + sent + ") message id " + message.getHeaders().getId());
logger.trace("postSend (sent=" + sent + ")");
}
for (ChannelInterceptor interceptor : this.interceptors) {
interceptor.postSend(message, channel, sent);

20
spring-messaging/src/main/java/org/springframework/messaging/support/ErrorMessage.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,8 @@
package org.springframework.messaging.support;
import org.springframework.messaging.MessageHeaders;
import java.util.Map;
/**
@@ -43,12 +45,26 @@ public class ErrorMessage extends GenericMessage<Throwable> {
/**
* Create a new message with the given payload and headers.
* The content of the given header map is copied.
*
* @param payload the message payload, never {@code null}
* @param headers message headers
* @param headers message headers to use for initialization
*/
public ErrorMessage(Throwable payload, Map<String, Object> headers) {
super(payload, headers);
}
/**
* A constructor with the {@link MessageHeaders} instance to use.
*
* <p><strong>Note:</strong> the given {@code MessageHeaders} instance is used
* directly in the new message, i.e. it is not copied.
*
* @param payload the message payload, never {@code null}
* @param headers message headers
*/
public ErrorMessage(Throwable payload, MessageHeaders headers) {
super(payload, headers);
}
}

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

@@ -49,18 +49,33 @@ public class GenericMessage<T> implements Message<T>, Serializable {
* @param payload the message payload, never {@code null}
*/
public GenericMessage(T payload) {
this(payload, null);
this(payload, new MessageHeaders(null));
}
/**
* Create a new message with the given payload and headers.
* The content of the given header map is copied.
*
* @param payload the message payload, never {@code null}
* @param headers message headers to use for initialization
*/
public GenericMessage(T payload, Map<String, Object> headers) {
this(payload, new MessageHeaders(headers));
}
/**
* A constructor with the {@link MessageHeaders} instance to use.
*
* <p><strong>Note:</strong> the given {@code MessageHeaders} instance is used
* directly in the new message, i.e. it is not copied.
*
* @param payload the message payload, never {@code null}
* @param headers message headers
*/
public GenericMessage(T payload, Map<String, Object> headers) {
public GenericMessage(T payload, MessageHeaders headers) {
Assert.notNull(headers, "'headers' must not be null");
Assert.notNull(payload, "payload must not be null");
this.headers = new MessageHeaders(headers);
this.headers = headers;
this.payload = payload;
}
@@ -96,7 +111,7 @@ public class GenericMessage<T> implements Message<T>, Serializable {
}
if (obj != null && obj instanceof GenericMessage<?>) {
GenericMessage<?> other = (GenericMessage<?>) obj;
return (this.headers.getId().equals(other.headers.getId()) &&
return (ObjectUtils.nullSafeEquals(this.headers.getId(), other.headers.getId()) &&
this.headers.equals(other.headers) && this.payload.equals(other.payload));
}
return false;

74
spring-messaging/src/main/java/org/springframework/messaging/support/MessageBuilder.java
View File

@@ -20,6 +20,7 @@ import java.util.Map;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
/**
@@ -36,23 +37,27 @@ import org.springframework.util.Assert;
*/
public final class MessageBuilder<T> {
private final Message<T> originalMessage;
private final T payload;
private MessageHeaderAccessor headerAccessor;
private final Message<T> originalMessage;
/**
* Private constructor to be invoked from the static factory methods only.
*/
private MessageBuilder(T payload, Message<T> originalMessage) {
Assert.notNull(payload, "payload must not be null");
this.payload = payload;
this.originalMessage = originalMessage;
private MessageBuilder(Message<T> originalMessage) {
Assert.notNull(originalMessage, "'originalMessage' must not be null");
this.payload = originalMessage.getPayload();
this.headerAccessor = new MessageHeaderAccessor(originalMessage);
this.originalMessage = originalMessage;
}
private MessageBuilder(T payload, MessageHeaderAccessor accessor) {
Assert.notNull(payload, "'payload' must not be null");
Assert.notNull(accessor, "'messageHeaderAccessor' must not be null");
this.payload = payload;
this.headerAccessor = accessor;
this.originalMessage = null;
}
/**
* Create a builder for a new {@link Message} instance pre-populated with all of the
@@ -62,26 +67,49 @@ public final class MessageBuilder<T> {
* @param message the Message from which the payload and all headers will be copied
*/
public static <T> MessageBuilder<T> fromMessage(Message<T> message) {
Assert.notNull(message, "message must not be null");
return new MessageBuilder<T>(message.getPayload(), message);
return new MessageBuilder<T>(message);
}
/**
* Create a builder for a new {@link Message} instance with the provided payload.
* @param payload the payload for the new message
* Create a new builder for a message with the given payload.
* @param payload the payload
*/
public static <T> MessageBuilder<T> withPayload(T payload) {
return new MessageBuilder<T>(payload, null);
return new MessageBuilder<T>(payload, new MessageHeaderAccessor());
}
/**
* A shortcut factory method for creating a message with the given payload
* and {@code MessageHeaders}.
*
* <p><strong>Note:</strong> the given {@code MessageHeaders} instance is used
* directly in the new message, i.e. it is not copied.
*
* @param payload the payload to use, never {@code null}
* @param messageHeaders the headers to use, never {@code null}
* @return the created message
* @since 4.1
*/
@SuppressWarnings("unchecked")
public static <T> Message<T> createMessage(T payload, MessageHeaders messageHeaders) {
Assert.notNull(payload, "'payload' must not be null");
Assert.notNull(messageHeaders, "'messageHeaders' must not be null");
if (payload instanceof Throwable) {
return (Message<T>) new ErrorMessage((Throwable) payload, messageHeaders);
}
else {
return new GenericMessage<T>(payload, messageHeaders);
}
}
/**
* Set the message headers.
* @param headerAccessor the headers for the message
* Set the message headers to use by providing a {@code MessageHeaderAccessor}.
*
* @param accessor the headers to use
*/
public MessageBuilder<T> setHeaders(MessageHeaderAccessor headerAccessor) {
Assert.notNull(headerAccessor, "HeaderAccessor must not be null");
this.headerAccessor = headerAccessor;
public MessageBuilder<T> setHeaders(MessageHeaderAccessor accessor) {
Assert.notNull(accessor, "HeaderAccessor must not be null");
this.headerAccessor = accessor;
return this;
}
@@ -161,13 +189,17 @@ public final class MessageBuilder<T> {
@SuppressWarnings("unchecked")
public Message<T> build() {
if ((this.originalMessage != null) && !this.headerAccessor.isModified()) {
if (this.originalMessage != null && !this.headerAccessor.isModified()) {
return this.originalMessage;
}
if (this.payload instanceof Throwable) {
return (Message<T>) new ErrorMessage((Throwable) this.payload, this.headerAccessor.toMap());
}
return new GenericMessage<T>(this.payload, this.headerAccessor.toMap());
else {
return new GenericMessage<T>(this.payload, this.headerAccessor.toMap());
}
}
}

362
spring-messaging/src/main/java/org/springframework/messaging/support/MessageHeaderAccessor.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.
@@ -29,19 +29,74 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
import org.springframework.util.IdGenerator;
import org.springframework.util.MimeType;
import org.springframework.util.ObjectUtils;
import org.springframework.util.PatternMatchUtils;
import org.springframework.util.StringUtils;
/**
* A base class for read/write access to {@link MessageHeaders}. Supports creation of new
* headers or modification of existing message headers.
* A base for classes providing strongly typed getters and setters as well as
* behavior around specific categories of headers (e.g. STOMP headers).
* Supports creating new headers (default constructor), modifying existing headers
* (when still mutable), or copying and modifying existing headers.
*
* <p>Sub-classes can provide additional typed getters and setters for convenient access
* to specific headers. Getters and setters should delegate to {@link #getHeader(String)}
* or {@link #setHeader(String, Object)} respectively. At the end {@link #toMap()} can be
* used to obtain the resulting headers.
* <p>The {@link #getMessageHeaders()} method provides access to the underlying,
* fully-prepared {@code MessageHeaders} instance that can then be used as-is
* to create a single message as follows:
*
* <pre class="code">
* MessageHeaderAccessor headerAccessor = new MessageHeaderAccessor();
* headerAccessor.set("foo", "bar");
* Message message = MessageBuilder.createMessage("payload", headerAccessor.getMessageHeaders());
* </pre>
*
* <p>After the above is executed, by default the {@code MessageHeaderAccessor}
* is immutable. It is also possible to leave it mutable, for example for further
* initialization of the message in the same thread:
*
* <pre class="code">
* MessageHeaderAccessor headerAccessor = new MessageHeaderAccessor();
* headerAccessor.set("foo", "bar");
* headerAccessor.setLeaveMutable(true);
* Message message = MessageBuilder.createMessage("payload", headerAccessor.getMessageHeaders());
*
* // later on in the same thread...
*
* MessageHeaderAccessor headerAccessor = MessageHeaderAccessor.getAccessor(message);
* headerAccessor.set("bar", "baz");
* headerAccessor.setImmutable();
* </pre>
*
* <p>To re-obtain the {@code MessageHeaderAccessor} for a {@code MessageHeaders}
* instance, use the following:
*
* <pre class="code">
* MessageHeaderAccessor headerAccessor = new MessageHeaderAccessor();
* headerAccessor.set("foo", "bar");
* Message message = MessageBuilder.createMessage("payload", headerAccessor.getMessageHeaders());
*
* // later on (any thread)...
* MessageHeaderAccessor headerAccessor = MessageHeaderAccessor.getAccessor(message);
* headerAccessor.get("foo");
* </pre>
*
* <p>To prepare multiple messages with the same {@code MessageHeaderAccessor}
* instance, use the code below. However note that this usage style does not allow
* re-obtaining a header accessor later on:
* <pre class="code">
* MessageHeaderAccessor headerAccessor = new MessageHeaderAccessor();
* MessageBuilder builder = MessageBuilder.withPayload("payload").setHeaders(headerAccessor);
*
* headerAccessor.setHeader("foo", "bar1");
* Message message1 = builder.build();
*
* headerAccessor.setHeader("foo", "bar2");
* Message message2 = builder.build();
*
* headerAccessor.setHeader("foo", "bar3");
* Message message3 = builder.build();
* </pre>
*
* @author Rossen Stoyanchev
* @since 4.0
@@ -50,62 +105,209 @@ public class MessageHeaderAccessor {
protected Log logger = LogFactory.getLog(getClass());
private final MutableMessageHeaders headers;
// wrapped read-only message headers
private final MessageHeaders originalHeaders;
private boolean modified;
// header updates
private final Map<String, Object> headers = new HashMap<String, Object>(4);
private boolean leaveMutable;
private IdGenerator idGenerator;
private boolean enableTimestamp = false;
/**
* A constructor for creating new message headers.
* A constructor to create new headers.
*/
public MessageHeaderAccessor() {
this.originalHeaders = null;
this.headers = new MutableMessageHeaders();
}
/**
* A constructor for accessing and modifying existing message headers.
* A constructor accepting the headers of an existing message to copy.
*/
public MessageHeaderAccessor(Message<?> message) {
this.originalHeaders = (message != null) ? message.getHeaders() : null;
if (message != null) {
this.headers = new MutableMessageHeaders(message.getHeaders());
MessageHeaderAccessor accessor = getAccessor(message, MessageHeaderAccessor.class);
if (accessor != null) {
this.idGenerator = accessor.idGenerator;
this.enableTimestamp = accessor.enableTimestamp;
}
}
else {
this.headers = new MutableMessageHeaders();
}
}
/**
* Return the original {@code MessageHeaderAccessor} used to create the headers
* of the given {@code Message}, or {@code null} if that's not available or if
* its type does not match the required type.
*
* <p>This is for cases where the existence of an accessor is strongly expected
* (to be followed up with an assertion) or will created if not provided.
*
* @return an accessor instance of the specified type or {@code null}.
* @since 4.1
*/
@SuppressWarnings("unchecked")
public static <T extends MessageHeaderAccessor> T getAccessor(Message<?> message, Class<T> requiredType) {
return getAccessor(message.getHeaders(), requiredType);
}
/**
* Return a header map including original, wrapped headers (if any) plus additional
* header updates made through accessor methods.
* A variation of {@link #getAccessor(org.springframework.messaging.Message, Class)}
* with a {@code MessageHeaders} instance instead of a {@code Message}.
*
* <p>This is for cases when a full message may not have been created yet.
*
* @return an accessor instance of the specified type or {@code null}.
* @since 4.1
*/
@SuppressWarnings("unchecked")
public static <T extends MessageHeaderAccessor> T getAccessor(MessageHeaders messageHeaders, Class<T> requiredType) {
if (messageHeaders instanceof MutableMessageHeaders) {
MutableMessageHeaders mutableHeaders = (MutableMessageHeaders) messageHeaders;
MessageHeaderAccessor headerAccessor = mutableHeaders.getMessageHeaderAccessor();
if (requiredType.isAssignableFrom(headerAccessor.getClass())) {
return (T) headerAccessor;
}
}
return null;
}
/**
* Return a mutable {@code MessageHeaderAccessor} for the given message attempting
* to match the type of accessor used to create the message headers, or otherwise
* wrapping the message with a {@code MessageHeaderAccessor} instance.
*
* <p>This is for cases where a header needs to be updated in generic code
* while preserving the accessor type for downstream processing.
*
* @return an accessor of the required type, never {@code null}.
* @since 4.1
*/
public static MessageHeaderAccessor getMutableAccessor(Message<?> message) {
if (message.getHeaders() instanceof MutableMessageHeaders) {
MutableMessageHeaders mutableHeaders = (MutableMessageHeaders) message.getHeaders();
MessageHeaderAccessor accessor = mutableHeaders.getMessageHeaderAccessor();
if (accessor != null) {
return (accessor.isMutable() ? accessor : accessor.createAccessor(message));
}
}
return new MessageHeaderAccessor(message);
}
protected MessageHeaderAccessor createAccessor(Message<?> message) {
return new MessageHeaderAccessor(message);
}
/**
* Return the underlying {@code MessageHeaders} instance.
*
* <p>Unless {@link #setLeaveMutable(boolean)} was set to {@code true}, after
* this call, the headers are immutable and this accessor can no longer
* modify them.
*
* <p>This method always returns the same {@code MessageHeaders} instance if
* invoked multiples times. To obtain a copy of the underlying headers instead
* use {@link #toMap()}.
*/
public MessageHeaders getMessageHeaders() {
this.headers.setIdAndTimestamp();
if (!this.leaveMutable) {
setImmutable();
}
return this.headers;
}
/**
* Return a copy of the underlying header values.
*
* <p>This method can be invoked many times, with modifications in between
* where each new call returns a fresh copy of the current header values.
*/
public Map<String, Object> toMap() {
Map<String, Object> result = new HashMap<String, Object>();
if (this.originalHeaders != null) {
result.putAll(this.originalHeaders);
}
for (String key : this.headers.keySet()) {
Object value = this.headers.get(key);
if (value == null) {
result.remove(key);
}
else {
result.put(key, value);
}
}
return result;
return new HashMap<String, Object>(this.headers);
}
/**
* By default when {@link #getMessageHeaders()} is called, {@code "this"}
* {@code MessageHeaderAccessor} instance can no longer be used to modify the
* underlying message headers and the returned {@code MessageHeaders} is immutable.
*
* <p>However when this is set to {@code true}, the returned (underlying)
* {@code MessageHeaders} instance remains mutable. To make further modifications
* continue to use the same accessor instance or re-obtain it via:<br>
* {@link org.springframework.messaging.support.MessageHeaderAccessor#getAccessor(org.springframework.messaging.Message, Class)
* MessageHeaderAccessor.getAccessor(Message, Class)}
*
* <p>When modifications are complete use {@link #setImmutable()} to prevent
* further changes. The intended use case for this mechanism is initialization
* of a Message within a single thread.
*
* <p>By default this is set to {@code false}.
* @since 4.1
*/
public void setLeaveMutable(boolean leaveMutable) {
Assert.state(this.headers.isMutable(), "Already immutable");
this.leaveMutable = leaveMutable;
}
/**
* By default when {@link #getMessageHeaders()} is called, {@code "this"}
* {@code MessageHeaderAccessor} instance can no longer be used to modify the
* underlying message headers. However if {@link #setLeaveMutable(boolean)}
* is used, this method is necessary to indicate explicitly when the
* {@code MessageHeaders} instance should no longer be modified.
* @since 4.1
*/
public void setImmutable() {
this.headers.setImmutable();
}
/**
* Whether the underlying headers can still be modified.
* @since 4.1
*/
public boolean isMutable() {
return this.headers.isMutable();
}
/**
* A private mechanism for providing an alternate IdGenerator strategy.
*
* <p>By default this property is not set in which case the default IdGenerator
* of {@link org.springframework.messaging.MessageHeaders} is used.
*
* @see org.springframework.messaging.support.MessageHeaderAccessorFactorySupport
*/
void setIdGenerator(IdGenerator idGenerator) {
this.idGenerator = idGenerator;
}
/**
* A private mechanism to enable having a timestamp added to every message.
*
* <p>By default this property is set to false.
*
* @see org.springframework.messaging.support.MessageHeaderAccessorFactorySupport
*/
void setEnableTimestamp(boolean enableTimestamp) {
this.enableTimestamp = enableTimestamp;
}
public boolean isModified() {
return (!this.headers.isEmpty());
return this.modified;
}
protected void setModified(boolean modified) {
this.modified = modified;
}
public Object getHeader(String headerName) {
if (this.headers.containsKey(headerName)) {
return this.headers.get(headerName);
}
else if (this.originalHeaders != null) {
return this.originalHeaders.get(headerName);
}
return null;
return this.headers.get(headerName);
}
/**
@@ -116,7 +318,22 @@ public class MessageHeaderAccessor {
Assert.isTrue(!isReadOnly(name), "The '" + name + "' header is read-only.");
verifyType(name, value);
if (!ObjectUtils.nullSafeEquals(value, getHeader(name))) {
this.headers.put(name, value);
this.modified = true;
if (value != null) {
this.headers.getRawHeaders().put(name, value);
}
else {
this.headers.getRawHeaders().remove(name);
}
}
}
protected void verifyType(String headerName, Object headerValue) {
if (headerName != null && headerValue != null) {
if (MessageHeaders.ERROR_CHANNEL.equals(headerName) || MessageHeaders.REPLY_CHANNEL.endsWith(headerName)) {
Assert.isTrue(headerValue instanceof MessageChannel || headerValue instanceof String, "The '"
+ headerName + "' header value must be a MessageChannel or String.");
}
}
}
@@ -144,7 +361,6 @@ public class MessageHeaderAccessor {
if (StringUtils.hasLength(pattern)){
if (pattern.contains("*")){
headersToRemove.addAll(getMatchingHeaderNames(pattern, this.headers));
headersToRemove.addAll(getMatchingHeaderNames(pattern, this.originalHeaders));
}
else {
headersToRemove.add(pattern);
@@ -251,17 +467,59 @@ public class MessageHeaderAccessor {
@Override
public String toString() {
return getClass().getSimpleName() + " [originalHeaders=" + this.originalHeaders
+ ", updated headers=" + this.headers + "]";
return getClass().getSimpleName() + " [headers=" + this.headers + "]";
}
@SuppressWarnings("serial")
private class MutableMessageHeaders extends MessageHeaders {
private boolean immutable;
public MutableMessageHeaders() {
this(null);
}
public MutableMessageHeaders(Map<String, Object> headers) {
super(headers, MessageHeaders.ID_VALUE_NONE, -1L);
}
public MessageHeaderAccessor getMessageHeaderAccessor() {
return MessageHeaderAccessor.this;
}
@Override
public Map<String, Object> getRawHeaders() {
Assert.state(!this.immutable, "Already immutable");
return super.getRawHeaders();
}
public void setImmutable() {
this.immutable = true;
}
public boolean isMutable() {
return !this.immutable;
}
public void setIdAndTimestamp() {
if (getId() == null) {
IdGenerator idGenerator = (MessageHeaderAccessor.this.idGenerator != null) ?
MessageHeaderAccessor.this.idGenerator :
MessageHeaders.getIdGenerator();
UUID id = idGenerator.generateId();
if (id != null && id != MessageHeaders.ID_VALUE_NONE) {
getRawHeaders().put(ID, id);
}
}
if (getTimestamp() == null) {
if (MessageHeaderAccessor.this.enableTimestamp) {
getRawHeaders().put(TIMESTAMP, System.currentTimeMillis());
}
}
}
}
protected void verifyType(String headerName, Object headerValue) {
if (headerName != null && headerValue != null) {
if (MessageHeaders.ERROR_CHANNEL.equals(headerName)
|| MessageHeaders.REPLY_CHANNEL.endsWith(headerName)) {
Assert.isTrue(headerValue instanceof MessageChannel || headerValue instanceof String, "The '"
+ headerName + "' header value must be a MessageChannel or String.");
}
}
}
}

67
spring-messaging/src/main/java/org/springframework/messaging/support/MessageHeaderAccessorFactorySupport.java Normal file
View File

@@ -0,0 +1,67 @@
/*
* 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.messaging.support;
import org.springframework.util.IdGenerator;
/**
* A support class for factories creating pre-configured instances of type
* {@link org.springframework.messaging.support.MessageHeaderAccessor}.
*
* @author Rossen Stoyanchev
* @since 4.1
*/
public class MessageHeaderAccessorFactorySupport {
private IdGenerator idGenerator;
private Boolean enableTimestamp;
protected MessageHeaderAccessorFactorySupport() {
}
/**
*
* @param idGenerator
*/
public void setIdGenerator(IdGenerator idGenerator) {
this.idGenerator = idGenerator;
}
public IdGenerator getIdGenerator() {
return this.idGenerator;
}
public void setEnableTimestamp(boolean enableTimestamp) {
this.enableTimestamp = enableTimestamp;
}
public boolean isEnableTimestamp() {
return this.enableTimestamp;
}
protected void updateMessageHeaderAccessor(MessageHeaderAccessor headerAccessor) {
if (this.idGenerator != null) {
headerAccessor.setIdGenerator(this.idGenerator);
}
if (this.enableTimestamp != null) {
headerAccessor.setEnableTimestamp(this.enableTimestamp);
}
}
}

170
spring-messaging/src/main/java/org/springframework/messaging/support/NativeMessageHeaderAccessor.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,25 +16,26 @@
package org.springframework.messaging.support;
import java.util.HashMap;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.util.ObjectUtils;
/**
* An extension of {@link MessageHeaderAccessor} that also stores and provides read/write
* access to message headers from an external source -- e.g. a Spring {@link Message}
* created to represent a STOMP message received from a STOMP client or message broker.
* Native message headers are kept in a {@link MultiValueMap} under the key
* Native message headers are kept in a {@code Map<String, List<String>>} under the key
* {@link #NATIVE_HEADERS}.
* <p>
* This class is not intended for direct use but is rather expected to be consumed
* through sub-classes such as
* This class is not intended for direct use but is rather expected to be used
* indirectly through protocol-specific sub-classes such as
* {@link org.springframework.messaging.simp.stomp.StompHeaderAccessor StompHeaderAccessor}.
* Such sub-classes may provide factory methods to translate message headers from
* an external messaging source (e.g. STOMP) to Spring {@link Message} headers and
@@ -49,107 +50,140 @@ public class NativeMessageHeaderAccessor extends MessageHeaderAccessor {
public static final String NATIVE_HEADERS = "nativeHeaders";
// wrapped native headers
private final Map<String, List<String>> originalNativeHeaders;
// native header updates
private final MultiValueMap<String, String> nativeHeaders = new LinkedMultiValueMap<String, String>(4);
/**
* A constructor for creating new headers, accepting an optional native header map.
* A protected constructor to create new headers.
*/
protected NativeMessageHeaderAccessor(Map<String, List<String>> nativeHeaders) {
this.originalNativeHeaders = nativeHeaders;
protected NativeMessageHeaderAccessor() {
this((Map<String, List<String>>) null);
}
/**
* A constructor for accessing and modifying existing message headers.
* A protected constructor to create new headers.
* @param nativeHeaders native headers to create the message with, may be {@code null}
*/
protected NativeMessageHeaderAccessor(Map<String, List<String>> nativeHeaders) {
if (!CollectionUtils.isEmpty(nativeHeaders)) {
setHeader(NATIVE_HEADERS, new LinkedMultiValueMap<String, String>(nativeHeaders));
}
}
/**
* A protected constructor accepting the headers of an existing message to copy.
*/
protected NativeMessageHeaderAccessor(Message<?> message) {
super(message);
this.originalNativeHeaders = initNativeHeaders(message);
}
private static Map<String, List<String>> initNativeHeaders(Message<?> message) {
if (message != null) {
@SuppressWarnings("unchecked")
Map<String, List<String>> headers = (Map<String, List<String>>) message.getHeaders().get(NATIVE_HEADERS);
if (headers != null) {
return headers;
Map<String, List<String>> map = (Map<String, List<String>>) getHeader(NATIVE_HEADERS);
if (map != null) {
// Force removal since setHeader checks for equality
removeHeader(NATIVE_HEADERS);
setHeader(NATIVE_HEADERS, new LinkedMultiValueMap<String, String>(map));
}
}
return null;
}
@Override
public Map<String, Object> toMap() {
Map<String, Object> result = super.toMap();
result.put(NATIVE_HEADERS, toNativeHeaderMap());
return result;
}
@Override
public boolean isModified() {
return (super.isModified() || (!this.nativeHeaders.isEmpty()));
@SuppressWarnings("unchecked")
private Map<String, List<String>> getNativeHeaders() {
return (Map<String, List<String>>) getHeader(NATIVE_HEADERS);
}
/**
* Return a map with native headers including original, wrapped headers (if any) plus
* additional header updates made through accessor methods.
* Return a copy of the native header values or an empty map.
*/
public Map<String, List<String>> toNativeHeaderMap() {
Map<String, List<String>> result = new HashMap<String, List<String>>();
if (this.originalNativeHeaders != null) {
result.putAll(this.originalNativeHeaders);
}
for (String key : this.nativeHeaders.keySet()) {
List<String> value = this.nativeHeaders.get(key);
if (value == null) {
result.remove(key);
}
else {
result.put(key, value);
Map<String, List<String>> map = getNativeHeaders();
return (map != null ? new LinkedMultiValueMap<String, String>(map) : Collections.<String, List<String>>emptyMap());
}
@Override
public void setImmutable() {
if (isMutable()) {
Map<String, List<String>> map = getNativeHeaders();
if (map != null) {
// Force removal since setHeader checks for equality
removeHeader(NATIVE_HEADERS);
setHeader(NATIVE_HEADERS, Collections.<String, List<String>>unmodifiableMap(map));
}
super.setImmutable();
}
return result;
}
/**
* Return all values for the specified native header or {@code null}.
* Whether the native header map contains the give header name.
*/
public boolean containsNativeHeader(String headerName) {
Map<String, List<String>> map = getNativeHeaders();
return (map != null ? map.containsKey(headerName) : false);
}
/**
* @return all values for the specified native header or {@code null}.
*/
public List<String> getNativeHeader(String headerName) {
if (this.nativeHeaders.containsKey(headerName)) {
return this.nativeHeaders.get(headerName);
}
else if (this.originalNativeHeaders != null) {
return this.originalNativeHeaders.get(headerName);
Map<String, List<String>> map = getNativeHeaders();
return (map != null ? map.get(headerName) : null);
}
/**
* @return the first value for the specified native header of {@code null}.
*/
public String getFirstNativeHeader(String headerName) {
Map<String, List<String>> map = getNativeHeaders();
if (map != null) {
List<String> values = map.get(headerName);
if (values != null) {
return values.get(0);
}
}
return null;
}
/**
* Return the first value for the specified native header of {@code null}.
*/
public String getFirstNativeHeader(String headerName) {
List<String> values = getNativeHeader(headerName);
return CollectionUtils.isEmpty(values) ? null : values.get(0);
}
/**
* Set the specified native header value.
* Set the specified native header value replacing existing values.
*/
public void setNativeHeader(String name, String value) {
if (!ObjectUtils.nullSafeEquals(value, getHeader(name))) {
this.nativeHeaders.set(name, value);
Assert.state(isMutable(), "Already immutable");
Map<String, List<String>> map = getNativeHeaders();
if (value == null) {
if (map != null && map.get(name) != null) {
setModified(true);
map.remove(name);
}
return;
}
if (map == null) {
map = new LinkedMultiValueMap<String, String>(4);
setHeader(NATIVE_HEADERS, map);
}
List<String> values = new LinkedList<String>();
values.add(value);
if (!ObjectUtils.nullSafeEquals(values, getHeader(name))) {
setModified(true);
map.put(name, values);
}
}
/**
* Add the specified native header value.
* Add the specified native header value to existing values.
*/
public void addNativeHeader(String name, String value) {
this.nativeHeaders.add(name, value);
Assert.state(isMutable(), "Already immutable");
if (value == null) {
return;
}
Map<String, List<String>> nativeHeaders = getNativeHeaders();
if (nativeHeaders == null) {
nativeHeaders = new LinkedMultiValueMap<String, String>(4);
setHeader(NATIVE_HEADERS, nativeHeaders);
}
List<String> values = nativeHeaders.get(name);
if (values == null) {
values = new LinkedList<String>();
nativeHeaders.put(name, values);
}
values.add(value);
setModified(true);
}
}