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

Introduce MessageHeader accessor types

Browse Source 
A new type MessageHeaderAccesssor provides read/write access to
MessageHeaders along with typed getter/setter methods along the lines
of the existing MessageBuilder methods (internally MessageBuilder
merely delegates to MessageHeaderAccessor). This class is extensible
with sub-classes expected to provide typed getter/setter methods for
specific categories of message headers.

NativeMessageHeaderAccessor is one specific sub-class that further
provides read/write access to headers from some external message
source (e.g. STOMP headers). Native headers are stored in a separate
MultiValueMap and kept under a specific key.
This commit is contained in:
Rossen Stoyanchev
2013-06-25 16:31:52 -04:00
parent ac23832e4d
commit 486b4101ec
19 changed files with 653 additions and 643 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-websocket/src/main/java/org/springframework/web/messaging/service/AbstractPubSubMessageHandler.java
View File

@@ -30,7 +30,7 @@ import org.springframework.util.AntPathMatcher;
import org.springframework.util.CollectionUtils;
import org.springframework.util.PathMatcher;
import org.springframework.web.messaging.MessageType;
import org.springframework.web.messaging.support.PubSubHeaderAccesssor;
import org.springframework.web.messaging.support.WebMessageHeaderAccesssor;
/**
@@ -80,7 +80,7 @@ public abstract class AbstractPubSubMessageHandler implements MessageHandler {
protected boolean isDestinationAllowed(Message<?> message) {
PubSubHeaderAccesssor headers = PubSubHeaderAccesssor.wrap(message);
WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.wrap(message);
String destination = headers.getDestination();
if (destination == null) {
@@ -116,7 +116,7 @@ public abstract class AbstractPubSubMessageHandler implements MessageHandler {
@Override
public final void handleMessage(Message<?> message) throws MessagingException {
PubSubHeaderAccesssor headers = PubSubHeaderAccesssor.wrap(message);
WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.wrap(message);
MessageType messageType = headers.getMessageType();
if (!canHandle(message, messageType)) {

12
spring-websocket/src/main/java/org/springframework/web/messaging/service/ReactorPubSubMessageHandler.java
View File

@@ -31,7 +31,7 @@ import org.springframework.web.messaging.MessageType;
import org.springframework.web.messaging.PubSubChannelRegistry;
import org.springframework.web.messaging.converter.CompositeMessageConverter;
import org.springframework.web.messaging.converter.MessageConverter;
import org.springframework.web.messaging.support.PubSubHeaderAccesssor;
import org.springframework.web.messaging.support.WebMessageHeaderAccesssor;
import reactor.core.Reactor;
import reactor.fn.Consumer;
@@ -78,7 +78,7 @@ public class ReactorPubSubMessageHandler extends AbstractPubSubMessageHandler {
logger.debug("Subscribe " + message);
}
PubSubHeaderAccesssor headers = PubSubHeaderAccesssor.wrap(message);
WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.wrap(message);
String subscriptionId = headers.getSubscriptionId();
BroadcastingConsumer consumer = new BroadcastingConsumer(subscriptionId);
@@ -107,7 +107,7 @@ public class ReactorPubSubMessageHandler extends AbstractPubSubMessageHandler {
try {
// Convert to byte[] payload before the fan-out
PubSubHeaderAccesssor headers = PubSubHeaderAccesssor.wrap(message);
WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.wrap(message);
byte[] payload = payloadConverter.convertToPayload(message.getPayload(), headers.getContentType());
Message<?> m = MessageBuilder.withPayload(payload).copyHeaders(message.getHeaders()).build();
@@ -120,7 +120,7 @@ public class ReactorPubSubMessageHandler extends AbstractPubSubMessageHandler {
@Override
public void handleDisconnect(Message<?> message) {
PubSubHeaderAccesssor headers = PubSubHeaderAccesssor.wrap(message);
WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.wrap(message);
removeSubscriptions(headers.getSessionId());
}
@@ -149,11 +149,11 @@ public class ReactorPubSubMessageHandler extends AbstractPubSubMessageHandler {
Message<?> sentMessage = event.getData();
PubSubHeaderAccesssor headers = PubSubHeaderAccesssor.wrap(sentMessage);
WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.wrap(sentMessage);
headers.setSubscriptionId(this.subscriptionId);
Message<?> clientMessage = MessageBuilder.withPayload(
sentMessage.getPayload()).copyHeaders(headers.toHeaders()).build();
sentMessage.getPayload()).copyHeaders(headers.toMap()).build();
clientChannel.send(clientMessage);
}

4
spring-websocket/src/main/java/org/springframework/web/messaging/service/method/AnnotationPubSubMessageHandler.java
View File

@@ -45,7 +45,7 @@ import org.springframework.web.messaging.annotation.UnsubscribeEvent;
import org.springframework.web.messaging.converter.MessageConverter;
import org.springframework.web.messaging.service.AbstractPubSubMessageHandler;
import org.springframework.web.messaging.support.MessageHolder;
import org.springframework.web.messaging.support.PubSubHeaderAccesssor;
import org.springframework.web.messaging.support.WebMessageHeaderAccesssor;
import org.springframework.web.method.HandlerMethod;
import org.springframework.web.method.HandlerMethodSelector;
@@ -185,7 +185,7 @@ public class AnnotationPubSubMessageHandler extends AbstractPubSubMessageHandler
private void handleMessageInternal(final Message<?> message, Map<MappingInfo, HandlerMethod> handlerMethods) {
PubSubHeaderAccesssor headers = PubSubHeaderAccesssor.wrap(message);
WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.wrap(message);
String destination = headers.getDestination();
HandlerMethod match = getHandlerMethod(destination, handlerMethods);

4
spring-websocket/src/main/java/org/springframework/web/messaging/service/method/MessageBodyArgumentResolver.java
View File

@@ -25,7 +25,7 @@ import org.springframework.web.messaging.annotation.MessageBody;
import org.springframework.web.messaging.converter.CompositeMessageConverter;
import org.springframework.web.messaging.converter.MessageConversionException;
import org.springframework.web.messaging.converter.MessageConverter;
import org.springframework.web.messaging.support.PubSubHeaderAccesssor;
import org.springframework.web.messaging.support.WebMessageHeaderAccesssor;
/**
@@ -52,7 +52,7 @@ public class MessageBodyArgumentResolver implements ArgumentResolver {
Object arg = null;
MessageBody annot = parameter.getParameterAnnotation(MessageBody.class);
MediaType contentType = (MediaType) message.getHeaders().get(PubSubHeaderAccesssor.CONTENT_TYPE);
MediaType contentType = (MediaType) message.getHeaders().get(WebMessageHeaderAccesssor.CONTENT_TYPE);
if (annot == null || annot.required()) {
Class<?> sourceType = message.getPayload().getClass();

8
spring-websocket/src/main/java/org/springframework/web/messaging/service/method/MessageReturnValueHandler.java
View File

@@ -21,7 +21,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.web.messaging.support.PubSubHeaderAccesssor;
import org.springframework.web.messaging.support.WebMessageHeaderAccesssor;
/**
@@ -66,10 +66,10 @@ public class MessageReturnValueHandler implements ReturnValueHandler {
return;
}
PubSubHeaderAccesssor headers = PubSubHeaderAccesssor.wrap(message);
WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.wrap(message);
Assert.notNull(headers.getSubscriptionId(), "No subscription id: " + message);
PubSubHeaderAccesssor returnHeaders = PubSubHeaderAccesssor.wrap(returnMessage);
WebMessageHeaderAccesssor returnHeaders = WebMessageHeaderAccesssor.wrap(returnMessage);
returnHeaders.setSessionId(headers.getSessionId());
returnHeaders.setSubscriptionId(headers.getSubscriptionId());
@@ -78,7 +78,7 @@ public class MessageReturnValueHandler implements ReturnValueHandler {
}
returnMessage = MessageBuilder.withPayload(
returnMessage.getPayload()).copyHeaders(returnHeaders.toHeaders()).build();
returnMessage.getPayload()).copyHeaders(returnHeaders.toMap()).build();
this.clientChannel.send(returnMessage);
}

147
spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompHeaderAccessor.java
View File

@@ -16,8 +16,8 @@
package org.springframework.web.messaging.stomp.support;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -26,12 +26,9 @@ import java.util.concurrent.atomic.AtomicLong;
import org.springframework.http.MediaType;
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.StringUtils;
import org.springframework.web.messaging.stomp.StompCommand;
import org.springframework.web.messaging.support.PubSubHeaderAccesssor;
import org.springframework.web.messaging.support.WebMessageHeaderAccesssor;
/**
@@ -39,13 +36,13 @@ import org.springframework.web.messaging.support.PubSubHeaderAccesssor;
* STOMP-specific headers of an existing message.
* <p>
* Use one of the static factory method in this class, then call getters and setters, and
* at the end if necessary call {@link #toHeaders()} to obtain the updated headers
* or call {@link #toStompMessageHeaders()} to obtain only the STOMP-specific headers.
* at the end if necessary call {@link #toMap()} to obtain the updated headers
* or call {@link #toNativeHeaderMap()} to obtain only the STOMP-specific headers.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
public class StompHeaderAccessor extends PubSubHeaderAccesssor {
public class StompHeaderAccessor extends WebMessageHeaderAccesssor {
public static final String STOMP_ID = "id";
@@ -80,55 +77,40 @@ public class StompHeaderAccessor extends PubSubHeaderAccesssor {
public static final String HEARTBEAT = "heart-beat";
private static final String STOMP_HEADERS = "stompHeaders";
private static final AtomicLong messageIdCounter = new AtomicLong();
private final Map<String, String> headers;
/**
* A constructor for creating new STOMP message headers.
* This constructor is private. See factory methods in this sub-classes.
*/
private StompHeaderAccessor(StompCommand command, Map<String, List<String>> externalSourceHeaders) {
super(command.getMessageType(), command, externalSourceHeaders);
this.headers = new HashMap<String, String>(4);
updateMessageHeaders();
initWebMessageHeaders();
}
private void updateMessageHeaders() {
if (getExternalSourceHeaders().isEmpty()) {
return;
}
String destination = getHeaderValue(DESTINATION);
private void initWebMessageHeaders() {
String destination = getFirstNativeHeader(DESTINATION);
if (destination != null) {
super.setDestination(destination);
}
String contentType = getHeaderValue(CONTENT_TYPE);
String contentType = getFirstNativeHeader(CONTENT_TYPE);
if (contentType != null) {
super.setContentType(MediaType.parseMediaType(contentType));
}
if (StompCommand.SUBSCRIBE.equals(getStompCommand())) {
if (getHeaderValue(STOMP_ID) != null) {
super.setSubscriptionId(getHeaderValue(STOMP_ID));
if (getFirstNativeHeader(STOMP_ID) != null) {
super.setSubscriptionId(getFirstNativeHeader(STOMP_ID));
}
}
}
/**
* A constructor for accessing and modifying existing message headers. This
* constructor is protected. See factory methods in this class.
* A constructor for accessing and modifying existing message headers.
*/
@SuppressWarnings("unchecked")
private StompHeaderAccessor(Message<?> message) {
super(message);
this.headers = (message.getHeaders() .get(STOMP_HEADERS) != null) ?
(Map<String, String>) message.getHeaders().get(STOMP_HEADERS) : new HashMap<String, String>(4);
}
/**
* Create {@link StompHeaderAccessor} for a new {@link Message}.
*/
@@ -152,53 +134,35 @@ public class StompHeaderAccessor extends PubSubHeaderAccesssor {
/**
* Return the original, wrapped headers (i.e. unmodified) or a new Map including any
* updates made via setters.
* Return STOMP headers including original, wrapped STOMP headers (if any) plus
* additional header updates made through accessor methods.
*/
@Override
public Map<String, Object> toHeaders() {
Map<String, Object> result = super.toHeaders();
if (isModified()) {
result.put(STOMP_HEADERS, this.headers);
}
return result;
}
public Map<String, List<String>> toNativeHeaderMap() {
@Override
public boolean isModified() {
return (super.isModified() || !this.headers.isEmpty());
}
/**
* Return STOMP headers and any custom headers that may have been sent by
* a remote endpoint, if this message originated from outside.
*/
public Map<String, List<String>> toStompMessageHeaders() {
MultiValueMap<String, String> result = new LinkedMultiValueMap<String, String>();
result.putAll(getExternalSourceHeaders());
result.setAll(this.headers);
Map<String, List<String>> result = super.toNativeHeaderMap();
String destination = super.getDestination();
if (destination != null) {
result.set(DESTINATION, destination);
result.put(DESTINATION, Arrays.asList(destination));
}
MediaType contentType = getContentType();
if (contentType != null) {
result.set(CONTENT_TYPE, contentType.toString());
result.put(CONTENT_TYPE, Arrays.asList(contentType.toString()));
}
if (StompCommand.MESSAGE.equals(getStompCommand())) {
String subscriptionId = getSubscriptionId();
if (subscriptionId != null) {
result.set(SUBSCRIPTION, subscriptionId);
result.put(SUBSCRIPTION, Arrays.asList(subscriptionId));
}
else {
logger.warn("STOMP MESSAGE frame should have a subscription: " + this.toString());
}
if ((getMessageId() == null)) {
result.set(MESSAGE_ID, getSessionId() + "-" + messageIdCounter.getAndIncrement());
String messageId = getSessionId() + "-" + messageIdCounter.getAndIncrement();
result.put(MESSAGE_ID, Arrays.asList(messageId));
}
}
@@ -216,42 +180,37 @@ public class StompHeaderAccessor extends PubSubHeaderAccesssor {
}
public Set<String> getAcceptVersion() {
String rawValue = getHeaderValue(ACCEPT_VERSION);
String rawValue = getFirstNativeHeader(ACCEPT_VERSION);
return (rawValue != null) ? StringUtils.commaDelimitedListToSet(rawValue) : Collections.<String>emptySet();
}
private String getHeaderValue(String headerName) {
List<String> values = getExternalSourceHeaders().get(headerName);
return !CollectionUtils.isEmpty(values) ? values.get(0) : this.headers.get(headerName);
}
public void setAcceptVersion(String acceptVersion) {
this.headers.put(ACCEPT_VERSION, acceptVersion);
setNativeHeader(ACCEPT_VERSION, acceptVersion);
}
public void setHost(String host) {
this.headers.put(HOST, host);
setNativeHeader(HOST, host);
}
public String getHost() {
return getHeaderValue(HOST);
return getFirstNativeHeader(HOST);
}
@Override
public void setDestination(String destination) {
super.setDestination(destination);
this.headers.put(DESTINATION, destination);
setNativeHeader(DESTINATION, destination);
}
@Override
public void setDestinations(List<String> destinations) {
Assert.isTrue((destinations != null) && (destinations.size() == 1), "STOMP allows one destination per message");
super.setDestinations(destinations);
this.headers.put(DESTINATION, destinations.get(0));
setNativeHeader(DESTINATION, destinations.get(0));
}
public long[] getHeartbeat() {
String rawValue = getHeaderValue(HEARTBEAT);
String rawValue = getFirstNativeHeader(HEARTBEAT);
if (!StringUtils.hasText(rawValue)) {
return null;
}
@@ -263,99 +222,91 @@ public class StompHeaderAccessor extends PubSubHeaderAccesssor {
public void setContentType(MediaType mediaType) {
if (mediaType != null) {
super.setContentType(mediaType);
this.headers.put(CONTENT_TYPE, mediaType.toString());
setNativeHeader(CONTENT_TYPE, mediaType.toString());
}
}
public MediaType getContentType() {
String value = getHeaderValue(CONTENT_TYPE);
String value = getFirstNativeHeader(CONTENT_TYPE);
return (value != null) ? MediaType.parseMediaType(value) : null;
}
public Integer getContentLength() {
String contentLength = getHeaderValue(CONTENT_LENGTH);
String contentLength = getFirstNativeHeader(CONTENT_LENGTH);
return StringUtils.hasText(contentLength) ? new Integer(contentLength) : null;
}
public void setContentLength(int contentLength) {
this.headers.put(CONTENT_LENGTH, String.valueOf(contentLength));
setNativeHeader(CONTENT_LENGTH, String.valueOf(contentLength));
}
public void setHeartbeat(long cx, long cy) {
this.headers.put(HEARTBEAT, StringUtils.arrayToCommaDelimitedString(new Object[] {cx, cy}));
setNativeHeader(HEARTBEAT, StringUtils.arrayToCommaDelimitedString(new Object[] {cx, cy}));
}
public void setAck(String ack) {
this.headers.put(ACK, ack);
setNativeHeader(ACK, ack);
}
public String getAck() {
return getHeaderValue(ACK);
return getFirstNativeHeader(ACK);
}
public void setNack(String nack) {
this.headers.put(NACK, nack);
setNativeHeader(NACK, nack);
}
public String getNack() {
return getHeaderValue(NACK);
return getFirstNativeHeader(NACK);
}
public void setLogin(String login) {
this.headers.put(LOGIN, login);
setNativeHeader(LOGIN, login);
}
public String getLogin() {
return getHeaderValue(LOGIN);
return getFirstNativeHeader(LOGIN);
}
public void setPasscode(String passcode) {
this.headers.put(PASSCODE, passcode);
setNativeHeader(PASSCODE, passcode);
}
public String getPasscode() {
return getHeaderValue(PASSCODE);
return getFirstNativeHeader(PASSCODE);
}
public void setReceiptId(String receiptId) {
this.headers.put(RECEIPT_ID, receiptId);
setNativeHeader(RECEIPT_ID, receiptId);
}
public String getReceiptId() {
return getHeaderValue(RECEIPT_ID);
return getFirstNativeHeader(RECEIPT_ID);
}
public String getMessage() {
return getHeaderValue(MESSAGE);
return getFirstNativeHeader(MESSAGE);
}
public void setMessage(String content) {
this.headers.put(MESSAGE, content);
setNativeHeader(MESSAGE, content);
}
public String getMessageId() {
return getHeaderValue(MESSAGE_ID);
return getFirstNativeHeader(MESSAGE_ID);
}
public void setMessageId(String id) {
this.headers.put(MESSAGE_ID, id);
setNativeHeader(MESSAGE_ID, id);
}
public String getVersion() {
return getHeaderValue(VERSION);
return getFirstNativeHeader(VERSION);
}
public void setVersion(String version) {
this.headers.put(VERSION, version);
}
@Override
public String toString() {
return "StompHeaders [" + "messageType=" + getMessageType() + ", protocolMessageType="
+ getProtocolMessageType() + ", destination=" + getDestination()
+ ", subscriptionId=" + getSubscriptionId() + ", sessionId=" + getSessionId()
+ ", externalSourceHeaders=" + getExternalSourceHeaders() + ", headers=" + this.headers + "]";
setNativeHeader(VERSION, version);
}
}

4
spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompMessageConverter.java
View File

@@ -99,7 +99,7 @@ public class StompMessageConverter {
byte[] payload = new byte[totalLength - payloadIndex];
System.arraycopy(byteContent, payloadIndex, payload, 0, totalLength - payloadIndex);
return MessageBuilder.withPayload(payload).copyHeaders(stompHeaders.toHeaders()).build();
return MessageBuilder.withPayload(payload).copyHeaders(stompHeaders.toMap()).build();
}
private int findIndexOfPayload(byte[] bytes) {
@@ -146,7 +146,7 @@ public class StompMessageConverter {
try {
out.write(stompHeaders.getStompCommand().toString().getBytes("UTF-8"));
out.write(LF);
for (Entry<String, List<String>> entry : stompHeaders.toStompMessageHeaders().entrySet()) {
for (Entry<String, List<String>> entry : stompHeaders.toNativeHeaderMap().entrySet()) {
String key = entry.getKey();
key = replaceAllOutbound(key);
for (String value : entry.getValue()) {

12
spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompRelayPubSubMessageHandler.java
View File

@@ -39,7 +39,7 @@ import org.springframework.web.messaging.converter.CompositeMessageConverter;
import org.springframework.web.messaging.converter.MessageConverter;
import org.springframework.web.messaging.service.AbstractPubSubMessageHandler;
import org.springframework.web.messaging.stomp.StompCommand;
import org.springframework.web.messaging.support.PubSubHeaderAccesssor;
import org.springframework.web.messaging.support.WebMessageHeaderAccesssor;
import reactor.core.Environment;
import reactor.core.Promise;
@@ -120,7 +120,7 @@ public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler
this.tcpClient = new TcpClient.Spec<String, String>(NettyTcpClient.class)
.using(new Environment())
.codec(new DelimitedCodec<String, String>((byte) 0, true, StandardCodecs.STRING_CODEC))
.connect("127.0.0.1", 61613)
.connect("127.0.0.1", 61616)
.get();
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.CONNECT);
@@ -129,7 +129,7 @@ public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler
headers.setPasscode("guest");
headers.setHeartbeat(0, 0);
Message<?> message = MessageBuilder.withPayload(
new byte[0]).copyHeaders(headers.toStompMessageHeaders()).build();
new byte[0]).copyHeaders(headers.toNativeHeaderMap()).build();
RelaySession session = new RelaySession(message, headers) {
@Override
@@ -206,7 +206,7 @@ public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler
@Override
public void handleOther(Message<?> message) {
StompCommand command = (StompCommand) message.getHeaders().get(PubSubHeaderAccesssor.PROTOCOL_MESSAGE_TYPE);
StompCommand command = (StompCommand) message.getHeaders().get(WebMessageHeaderAccesssor.PROTOCOL_MESSAGE_TYPE);
Assert.notNull(command, "Expected STOMP command: " + message.getHeaders());
forwardMessage(message, command);
}
@@ -326,7 +326,7 @@ public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.ERROR);
headers.setSessionId(sessionId);
headers.setMessage(errorText);
Message<?> errorMessage = MessageBuilder.withPayload(new byte[0]).copyHeaders(headers.toHeaders()).build();
Message<?> errorMessage = MessageBuilder.withPayload(new byte[0]).copyHeaders(headers.toMap()).build();
sendMessageToClient(errorMessage);
}
@@ -372,7 +372,7 @@ public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler
MediaType contentType = headers.getContentType();
byte[] payload = payloadConverter.convertToPayload(message.getPayload(), contentType);
Message<?> byteMessage = MessageBuilder.withPayload(payload).copyHeaders(headers.toHeaders()).build();
Message<?> byteMessage = MessageBuilder.withPayload(payload).copyHeaders(headers.toMap()).build();
if (logger.isTraceEnabled()) {
logger.trace("Forwarding message " + byteMessage);

12
spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompWebSocketHandler.java
View File

@@ -38,7 +38,7 @@ import org.springframework.web.messaging.converter.CompositeMessageConverter;
import org.springframework.web.messaging.converter.MessageConverter;
import org.springframework.web.messaging.stomp.StompCommand;
import org.springframework.web.messaging.stomp.StompConversionException;
import org.springframework.web.messaging.support.PubSubHeaderAccesssor;
import org.springframework.web.messaging.support.WebMessageHeaderAccesssor;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
@@ -159,7 +159,7 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement
// TODO: security
Message<?> connectedMessage = MessageBuilder.withPayload(EMPTY_PAYLOAD).copyHeaders(
connectedHeaders.toHeaders()).build();
connectedHeaders.toMap()).build();
byte[] bytes = getStompMessageConverter().fromMessage(connectedMessage);
session.sendMessage(new TextMessage(new String(bytes, Charset.forName("UTF-8"))));
}
@@ -195,7 +195,7 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.ERROR);
headers.setMessage(error.getMessage());
Message<?> message = MessageBuilder.withPayload(EMPTY_PAYLOAD).copyHeaders(headers.toHeaders()).build();
Message<?> message = MessageBuilder.withPayload(EMPTY_PAYLOAD).copyHeaders(headers.toMap()).build();
byte[] bytes = this.stompMessageConverter.fromMessage(message);
try {
@@ -209,9 +209,9 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
this.sessionInfos.remove(session.getId());
PubSubHeaderAccesssor headers = PubSubHeaderAccesssor.create(MessageType.DISCONNECT);
WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.create(MessageType.DISCONNECT);
headers.setSessionId(session.getId());
Message<?> message = MessageBuilder.withPayload(new byte[0]).copyHeaders(headers.toHeaders()).build();
Message<?> message = MessageBuilder.withPayload(new byte[0]).copyHeaders(headers.toMap()).build();
this.outputChannel.send(message);
}
@@ -264,7 +264,7 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement
}
try {
Message<?> byteMessage = MessageBuilder.withPayload(payload).copyHeaders(headers.toHeaders()).build();
Message<?> byteMessage = MessageBuilder.withPayload(payload).copyHeaders(headers.toMap()).build();
byte[] bytes = getStompMessageConverter().fromMessage(byteMessage);
session.sendMessage(new TextMessage(new String(bytes, Charset.forName("UTF-8"))));
}

251
spring-websocket/src/main/java/org/springframework/web/messaging/support/PubSubHeaderAccesssor.java
View File

@@ -1,251 +0,0 @@
/*
* Copyright 2002-2013 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.web.messaging.support;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.http.MediaType;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.web.messaging.MessageType;
/**
* A base class for working with message headers in Web, messaging protocols that support
* the publish-subscribe message pattern. Provides uniform access to specific values
* common across protocols such as a destination, message type (publish,
* subscribe/unsubscribe), session id, and others.
* <p>
* This class can be used to prepare headers for a new pub-sub message, or to access
* and/or modify headers of an existing message.
* <p>
* Use one of the static factory method in this class, then call getters and setters, and
* at the end if necessary call {@link #toHeaders()} to obtain the updated headers.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
public class PubSubHeaderAccesssor {
protected Log logger = LogFactory.getLog(getClass());
public static final String DESTINATIONS = "destinations";
public static final String CONTENT_TYPE = "contentType";
public static final String MESSAGE_TYPE = "messageType";
public static final String PROTOCOL_MESSAGE_TYPE = "protocolMessageType";
public static final String SESSION_ID = "sessionId";
public static final String SUBSCRIPTION_ID = "subscriptionId";
public static final String EXTERNAL_SOURCE_HEADERS = "extSourceHeaders";
private static final Map<String, List<String>> emptyMultiValueMap =
Collections.unmodifiableMap(new LinkedMultiValueMap<String, String>(0));
// wrapped read-only message headers
private final MessageHeaders originalHeaders;
// header updates
private final Map<String, Object> headers = new HashMap<String, Object>(4);
// saved headers from a message from a remote source
private final Map<String, List<String>> externalSourceHeaders;
/**
* A constructor for creating new message headers.
* This constructor is protected. See factory methods in this and sub-classes.
*/
protected PubSubHeaderAccesssor(MessageType messageType, Object protocolMessageType,
Map<String, List<String>> externalSourceHeaders) {
this.originalHeaders = null;
Assert.notNull(messageType, "messageType is required");
this.headers.put(MESSAGE_TYPE, messageType);
if (protocolMessageType != null) {
this.headers.put(PROTOCOL_MESSAGE_TYPE, protocolMessageType);
}
if (externalSourceHeaders == null) {
this.externalSourceHeaders = emptyMultiValueMap;
}
else {
this.externalSourceHeaders = Collections.unmodifiableMap(externalSourceHeaders); // TODO: list values must also be read-only
this.headers.put(EXTERNAL_SOURCE_HEADERS, this.externalSourceHeaders);
}
}
/**
* A constructor for accessing and modifying existing message headers. This
* constructor is protected. See factory methods in this and sub-classes.
*/
@SuppressWarnings("unchecked")
protected PubSubHeaderAccesssor(Message<?> message) {
Assert.notNull(message, "message is required");
this.originalHeaders = message.getHeaders();
this.externalSourceHeaders = (this.originalHeaders.get(EXTERNAL_SOURCE_HEADERS) != null) ?
(Map<String, List<String>>) this.originalHeaders.get(EXTERNAL_SOURCE_HEADERS) : emptyMultiValueMap;
}
/**
* Create {@link PubSubHeaderAccesssor} for a new {@link Message} with
* {@link MessageType#MESSAGE}.
*/
public static PubSubHeaderAccesssor create() {
return new PubSubHeaderAccesssor(MessageType.MESSAGE, null, null);
}
/**
* Create {@link PubSubHeaderAccesssor} for a new {@link Message} of a specific type.
*/
public static PubSubHeaderAccesssor create(MessageType messageType) {
return new PubSubHeaderAccesssor(messageType, null, null);
}
/**
* Create {@link PubSubHeaderAccesssor} from the headers of an existing message.
*/
public static PubSubHeaderAccesssor wrap(Message<?> message) {
return new PubSubHeaderAccesssor(message);
}
/**
* Return the original, wrapped headers (i.e. unmodified) or a new Map including any
* updates made via setters.
*/
public Map<String, Object> toHeaders() {
if (!isModified()) {
return this.originalHeaders;
}
Map<String, Object> result = new HashMap<String, Object>();
if (this.originalHeaders != null) {
result.putAll(this.originalHeaders);
}
result.putAll(this.headers);
return result;
}
public boolean isModified() {
return ((this.originalHeaders == null) || !this.headers.isEmpty());
}
public MessageType getMessageType() {
return (MessageType) getHeaderValue(MESSAGE_TYPE);
}
private Object getHeaderValue(String headerName) {
if (this.headers.get(headerName) != null) {
return this.headers.get(headerName);
}
else if ((this.originalHeaders != null) && (this.originalHeaders.get(headerName) != null)) {
return this.originalHeaders.get(headerName);
}
return null;
}
protected void setProtocolMessageType(Object protocolMessageType) {
this.headers.put(PROTOCOL_MESSAGE_TYPE, protocolMessageType);
}
protected Object getProtocolMessageType() {
return getHeaderValue(PROTOCOL_MESSAGE_TYPE);
}
public void setDestination(String destination) {
Assert.notNull(destination, "destination is required");
this.headers.put(DESTINATIONS, Arrays.asList(destination));
}
@SuppressWarnings("unchecked")
public String getDestination() {
List<String> destinations = (List<String>) getHeaderValue(DESTINATIONS);
return CollectionUtils.isEmpty(destinations) ? null : destinations.get(0);
}
@SuppressWarnings("unchecked")
public List<String> getDestinations() {
List<String> destinations = (List<String>) getHeaderValue(DESTINATIONS);
return CollectionUtils.isEmpty(destinations) ? null : destinations;
}
public void setDestinations(List<String> destinations) {
Assert.notNull(destinations, "destinations are required");
this.headers.put(DESTINATIONS, destinations);
}
public MediaType getContentType() {
return (MediaType) getHeaderValue(CONTENT_TYPE);
}
public void setContentType(MediaType contentType) {
Assert.notNull(contentType, "contentType is required");
this.headers.put(CONTENT_TYPE, contentType);
}
public String getSubscriptionId() {
return (String) getHeaderValue(SUBSCRIPTION_ID);
}
public void setSubscriptionId(String subscriptionId) {
this.headers.put(SUBSCRIPTION_ID, subscriptionId);
}
public String getSessionId() {
return (String) getHeaderValue(SESSION_ID);
}
public void setSessionId(String sessionId) {
this.headers.put(SESSION_ID, sessionId);
}
/**
* Return a read-only map of headers originating from a message received by the
* application from an external source (e.g. from a remote WebSocket endpoint). The
* header names and values are exactly as they were, and are protocol specific but may
* also be custom application headers if the protocol allows that.
*/
public Map<String, List<String>> getExternalSourceHeaders() {
return this.externalSourceHeaders;
}
@Override
public String toString() {
return "PubSubHeaders [originalHeaders=" + this.originalHeaders + ", headers="
+ this.headers + ", externalSourceHeaders=" + this.externalSourceHeaders + "]";
}
}

6
spring-websocket/src/main/java/org/springframework/web/messaging/support/PubSubMessageBuilder.java
View File

@@ -29,7 +29,7 @@ import reactor.util.Assert;
*/
public class PubSubMessageBuilder<T> {
private final PubSubHeaderAccesssor headers = PubSubHeaderAccesssor.create();
private final WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.create();
private final T payload;
@@ -67,11 +67,11 @@ public class PubSubMessageBuilder<T> {
Message<?> message = MessageHolder.getMessage();
if (message != null) {
String sessionId = PubSubHeaderAccesssor.wrap(message).getSessionId();
String sessionId = WebMessageHeaderAccesssor.wrap(message).getSessionId();
this.headers.setSessionId(sessionId);
}
return MessageBuilder.withPayload(this.payload).copyHeaders(this.headers.toHeaders()).build();
return MessageBuilder.withPayload(this.payload).copyHeaders(this.headers.toMap()).build();
}
}

167
spring-websocket/src/main/java/org/springframework/web/messaging/support/WebMessageHeaderAccesssor.java Normal file
View File

@@ -0,0 +1,167 @@
/*
* Copyright 2002-2013 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.web.messaging.support;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import org.springframework.http.MediaType;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.NativeMessageHeaderAccessor;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.web.messaging.MessageType;
/**
* A base class for working with message headers in Web, messaging protocols that support
* the publish-subscribe message pattern. Provides uniform access to specific values
* common across protocols such as a destination, message type (publish,
* subscribe/unsubscribe), session id, and others.
* <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.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
public class WebMessageHeaderAccesssor extends NativeMessageHeaderAccessor {
public static final String DESTINATIONS = "destinations";
public static final String CONTENT_TYPE = "contentType";
public static final String MESSAGE_TYPE = "messageType";
public static final String PROTOCOL_MESSAGE_TYPE = "protocolMessageType";
public static final String SESSION_ID = "sessionId";
public static final String SUBSCRIPTION_ID = "subscriptionId";
/**
* A constructor for creating new message headers.
* This constructor is protected. See factory methods in this and sub-classes.
*/
protected WebMessageHeaderAccesssor(MessageType messageType, Object protocolMessageType,
Map<String, List<String>> externalSourceHeaders) {
super(externalSourceHeaders);
Assert.notNull(messageType, "messageType is required");
setHeader(MESSAGE_TYPE, messageType);
if (protocolMessageType != null) {
setHeader(PROTOCOL_MESSAGE_TYPE, protocolMessageType);
}
}
/**
* A constructor for accessing and modifying existing message headers. This
* constructor is protected. See factory methods in this and sub-classes.
*/
protected WebMessageHeaderAccesssor(Message<?> message) {
super(message);
Assert.notNull(message, "message is required");
}
/**
* Create {@link WebMessageHeaderAccesssor} for a new {@link Message} with
* {@link MessageType#MESSAGE}.
*/
public static WebMessageHeaderAccesssor create() {
return new WebMessageHeaderAccesssor(MessageType.MESSAGE, null, null);
}
/**
* Create {@link WebMessageHeaderAccesssor} for a new {@link Message} of a specific type.
*/
public static WebMessageHeaderAccesssor create(MessageType messageType) {
return new WebMessageHeaderAccesssor(messageType, null, null);
}
/**
* Create {@link WebMessageHeaderAccesssor} from the headers of an existing message.
*/
public static WebMessageHeaderAccesssor wrap(Message<?> message) {
return new WebMessageHeaderAccesssor(message);
}
public MessageType getMessageType() {
return (MessageType) getHeader(MESSAGE_TYPE);
}
protected void setProtocolMessageType(Object protocolMessageType) {
setHeader(PROTOCOL_MESSAGE_TYPE, protocolMessageType);
}
protected Object getProtocolMessageType() {
return getHeader(PROTOCOL_MESSAGE_TYPE);
}
public void setDestination(String destination) {
Assert.notNull(destination, "destination is required");
setHeader(DESTINATIONS, Arrays.asList(destination));
}
@SuppressWarnings("unchecked")
public String getDestination() {
List<String> destinations = (List<String>) getHeader(DESTINATIONS);
return CollectionUtils.isEmpty(destinations) ? null : destinations.get(0);
}
@SuppressWarnings("unchecked")
public List<String> getDestinations() {
List<String> destinations = (List<String>) getHeader(DESTINATIONS);
return CollectionUtils.isEmpty(destinations) ? null : destinations;
}
public void setDestinations(List<String> destinations) {
Assert.notNull(destinations, "destinations are required");
setHeader(DESTINATIONS, destinations);
}
public MediaType getContentType() {
return (MediaType) getHeader(CONTENT_TYPE);
}
public void setContentType(MediaType contentType) {
Assert.notNull(contentType, "contentType is required");
setHeader(CONTENT_TYPE, contentType);
}
public String getSubscriptionId() {
return (String) getHeader(SUBSCRIPTION_ID);
}
public void setSubscriptionId(String subscriptionId) {
setHeader(SUBSCRIPTION_ID, subscriptionId);
}
public String getSessionId() {
return (String) getHeader(SESSION_ID);
}
public void setSessionId(String sessionId) {
setHeader(SESSION_ID, sessionId);
}
}

6
spring-websocket/src/test/java/org/springframework/web/messaging/stomp/support/StompMessageConverterTests.java
View File

@@ -53,7 +53,7 @@ public class StompMessageConverterTests {
MessageHeaders headers = message.getHeaders();
StompHeaderAccessor stompHeaders = StompHeaderAccessor.wrap(message);
assertEquals(7, stompHeaders.toHeaders().size());
assertEquals(7, stompHeaders.toMap().size());
assertEquals(Collections.singleton("1.1"), stompHeaders.getAcceptVersion());
assertEquals("github.org", stompHeaders.getHost());
@@ -84,7 +84,7 @@ public class StompMessageConverterTests {
StompHeaderAccessor stompHeaders = StompHeaderAccessor.wrap(message);
assertEquals(Collections.singleton("1.1"), stompHeaders.getAcceptVersion());
assertEquals("st\nomp.gi:thu\\b.org", stompHeaders.getExternalSourceHeaders().get("ho:\ns\rt").get(0));
assertEquals("st\nomp.gi:thu\\b.org", stompHeaders.toNativeHeaderMap().get("ho:\ns\rt").get(0));
String convertedBack = new String(this.converter.fromMessage(message), "UTF-8");
@@ -128,7 +128,7 @@ public class StompMessageConverterTests {
StompHeaderAccessor stompHeaders = StompHeaderAccessor.wrap(message);
assertEquals(Collections.singleton("1.1"), stompHeaders.getAcceptVersion());
assertEquals("st\nomp.gi:thu\\b.org", stompHeaders.getExternalSourceHeaders().get("ho:\ns\rt").get(0));
assertEquals("st\nomp.gi:thu\\b.org", stompHeaders.toNativeHeaderMap().get("ho:\ns\rt").get(0));
String convertedBack = new String(this.converter.fromMessage(message), "UTF-8");