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

Introduce CONNECT_ACK message type

Browse Source 
Previously, handling of a STOMP CONNECT message and sending of a
CONNECTED response was performed by StompProtocolHandler if it was
backed by SimpleBrokerMessageHandler, or left up to the real message
broker if it was backed by StompBrokerRelayMessageHandler. This
wasn't ideal as it should be StompProtocolHandler's job to simply map
messages to and from the STOMP protocol, not to do part of the
broker's job and respond directly to CONNECT.

This commit introduces a new message type, CONNECT_ACK. When it
receives a CONNECT message, SimpleBrokerMessageHandler will now
respond with a CONNECT_ACK message that StompProtocolHandler can map
into a STOMP CONNECTED message. The CONNECT_ACK message contains the
CONNECT message as a header so that StompProtocolHandler has access to
its accept-version header.

StompProtocolHandler has been simplified so that a CONNECT message
is always passed to the output channel, irrespective of whether it's
backed by a simple broker or a real broker. The handleConnect flag,
and the code that would set it correctly depending on the app's
configuration, has been removed.
This commit is contained in:
Andy Wilkinson
2013-09-27 15:41:46 +01:00
committed by Rossen Stoyanchev
parent b2f31a3c74
commit 5025c304b8
9 changed files with 90 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -41,6 +41,8 @@ import org.springframework.util.Assert;
*/
public class SimpMessageHeaderAccessor extends NativeMessageHeaderAccessor {
public static final String CONNECT_MESSAGE_HEADER = "connectMessage";
public static final String DESTINATION_HEADER = "destination";
public static final String MESSAGE_TYPE_HEADER = "messageType";

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

@@ -28,6 +28,8 @@ public enum SimpMessageType {
CONNECT,
CONNECT_ACK,
MESSAGE,
SUBSCRIBE,

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

@@ -54,8 +54,7 @@ public class ServletStompEndpointRegistry implements StompEndpointRegistry {
public ServletStompEndpointRegistry(WebSocketHandler webSocketHandler,
MutableUserQueueSuffixResolver userQueueSuffixResolver, TaskScheduler defaultSockJsTaskScheduler,
boolean handleConnect) {
MutableUserQueueSuffixResolver userQueueSuffixResolver, TaskScheduler defaultSockJsTaskScheduler) {
Assert.notNull(webSocketHandler);
Assert.notNull(userQueueSuffixResolver);
@@ -64,7 +63,6 @@ public class ServletStompEndpointRegistry implements StompEndpointRegistry {
this.subProtocolWebSocketHandler = findSubProtocolWebSocketHandler(webSocketHandler);
this.stompHandler = new StompProtocolHandler();
this.stompHandler.setUserQueueSuffixResolver(userQueueSuffixResolver);
this.stompHandler.setHandleConnect(handleConnect);
this.sockJsScheduler = defaultSockJsTaskScheduler;
}

3
spring-messaging/src/main/java/org/springframework/messaging/simp/config/WebSocketMessageBrokerConfigurationSupport.java
View File

@@ -57,9 +57,8 @@ public abstract class WebSocketMessageBrokerConfigurationSupport {
@Bean
public HandlerMapping brokerWebSocketHandlerMapping() {
boolean brokerRelayConfigured = getMessageBrokerConfigurer().getStompBrokerRelay() != null;
ServletStompEndpointRegistry registry = new ServletStompEndpointRegistry(subProtocolWebSocketHandler(),
userQueueSuffixResolver(), brokerDefaultSockJsTaskScheduler(), !brokerRelayConfigured);
userQueueSuffixResolver(), brokerDefaultSockJsTaskScheduler());
registerStompEndpoints(registry);
AbstractHandlerMapping hm = registry.getHandlerMapping();

13
spring-messaging/src/main/java/org/springframework/messaging/simp/handler/SimpleBrokerMessageHandler.java
View File

@@ -33,6 +33,8 @@ import org.springframework.util.MultiValueMap;
*/
public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
private static final byte[] EMPTY_PAYLOAD = new byte[0];
private final MessageChannel messageChannel;
private SubscriptionRegistry subscriptionRegistry = new DefaultSubscriptionRegistry();
@@ -96,8 +98,17 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
sendMessageToSubscribers(headers.getDestination(), message);
}
else if (SimpMessageType.DISCONNECT.equals(messageType)) {
String sessionId = SimpMessageHeaderAccessor.wrap(message).getSessionId();
String sessionId = headers.getSessionId();
this.subscriptionRegistry.unregisterAllSubscriptions(sessionId);
} else if (SimpMessageType.CONNECT.equals(messageType)) {
String sessionId = headers.getSessionId();
SimpMessageHeaderAccessor connectAckHeaders =
SimpMessageHeaderAccessor.create(SimpMessageType.CONNECT_ACK);
connectAckHeaders.setSessionId(sessionId);
connectAckHeaders.setHeader(SimpMessageHeaderAccessor.CONNECT_MESSAGE_HEADER, message);
Message<byte[]> connectAck =
MessageBuilder.withPayloadAndHeaders(EMPTY_PAYLOAD, connectAckHeaders).build();
this.messageChannel.send(connectAck);
}
}

96
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompProtocolHandler.java
View File

@@ -70,8 +70,6 @@ public class StompProtocolHandler implements SubProtocolHandler {
private MutableUserQueueSuffixResolver queueSuffixResolver = new SimpleUserQueueSuffixResolver();
private volatile boolean handleConnect = false;
/**
* Configure a resolver to use to maintain queue suffixes for user
* @see {@link org.springframework.messaging.simp.handler.UserDestinationMessageHandler}
@@ -87,29 +85,6 @@ public class StompProtocolHandler implements SubProtocolHandler {
return this.queueSuffixResolver;
}
/**
* Configures the handling of CONNECT frames. When {@code true}, CONNECT
* frames will be handled by this handler, and a CONNECTED response will be
* sent. When {@code false}, CONNECT frames will be forwarded for
* handling by another component.
*
* @param handleConnect {@code true} if connect frames should be handled
* by this handler, {@code false} otherwise.
*/
public void setHandleConnect(boolean handleConnect) {
this.handleConnect = handleConnect;
}
/**
* Returns whether or not this handler will handle CONNECT frames.
*
* @return Returns {@code true} if this handler will handle CONNECT frames,
* otherwise {@code false}.
*/
public boolean willHandleConnect() {
return this.handleConnect;
}
@Override
public List<String> getSupportedProtocols() {
return Arrays.asList("v10.stomp", "v11.stomp", "v12.stomp");
@@ -144,13 +119,7 @@ public class StompProtocolHandler implements SubProtocolHandler {
headers.setUser(session.getPrincipal());
message = MessageBuilder.withPayloadAndHeaders(message.getPayload(), headers).build();
if (this.handleConnect && SimpMessageType.CONNECT.equals(headers.getMessageType())) {
handleConnect(session, message);
}
else {
outputChannel.send(message);
}
outputChannel.send(message);
}
catch (Throwable t) {
logger.error("Terminating STOMP session due to failure to send message: ", t);
@@ -170,13 +139,15 @@ public class StompProtocolHandler implements SubProtocolHandler {
headers.setCommandIfNotSet(StompCommand.MESSAGE);
}
if (headers.getMessageType() == SimpMessageType.CONNECT_ACK) {
StompHeaderAccessor connectedHeaders = StompHeaderAccessor.create(StompCommand.CONNECTED);
connectedHeaders.setVersion(getVersion(headers));
connectedHeaders.setHeartbeat(0, 0);
headers = connectedHeaders;
}
if (headers.getCommand() == StompCommand.CONNECTED) {
if (this.handleConnect) {
// Ignore since we already sent it
return;
} else {
augmentConnectedHeaders(headers, session);
}
augmentConnectedHeaders(headers, session);
}
if (StompCommand.MESSAGE.equals(headers.getCommand()) && (headers.getSubscriptionId() == null)) {
@@ -208,35 +179,6 @@ public class StompProtocolHandler implements SubProtocolHandler {
}
}
protected void handleConnect(WebSocketSession session, Message<?> message) throws IOException {
StompHeaderAccessor connectHeaders = StompHeaderAccessor.wrap(message);
StompHeaderAccessor connectedHeaders = StompHeaderAccessor.create(StompCommand.CONNECTED);
Set<String> acceptVersions = connectHeaders.getAcceptVersion();
if (acceptVersions.contains("1.2")) {
connectedHeaders.setVersion("1.2");
}
else if (acceptVersions.contains("1.1")) {
connectedHeaders.setVersion("1.1");
}
else if (acceptVersions.isEmpty()) {
// 1.0
}
else {
throw new StompConversionException("Unsupported version '" + acceptVersions + "'");
}
connectedHeaders.setHeartbeat(0,0);
augmentConnectedHeaders(connectedHeaders, session);
// TODO: security
Message<byte[]> connectedMessage = MessageBuilder.withPayloadAndHeaders(new byte[0], connectedHeaders).build();
String payload = new String(this.stompEncoder.encode(connectedMessage), Charset.forName("UTF-8"));
session.sendMessage(new TextMessage(payload));
}
private void augmentConnectedHeaders(StompHeaderAccessor headers, WebSocketSession session) {
Principal principal = session.getPrincipal();
if (principal != null) {
@@ -287,4 +229,24 @@ public class StompProtocolHandler implements SubProtocolHandler {
outputChannel.send(message);
}
private String getVersion(StompHeaderAccessor connectAckHeaders) {
Message<?> connectMessage =
(Message<?>) connectAckHeaders.getHeader(StompHeaderAccessor.CONNECT_MESSAGE_HEADER);
StompHeaderAccessor connectHeaders = StompHeaderAccessor.wrap(connectMessage);
Set<String> acceptVersions = connectHeaders.getAcceptVersion();
if (acceptVersions.contains("1.2")) {
return "1.2";
}
else if (acceptVersions.contains("1.1")) {
return "1.1";
}
else if (acceptVersions.isEmpty()) {
return null;
}
else {
throw new StompConversionException("Unsupported version '" + acceptVersions + "'");
}
}
}