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

Add WebSocket transport configuration support

Browse Source 
Issue: SPR-11527
This commit is contained in:
Rossen Stoyanchev
2014-03-24 18:05:37 -04:00
parent 1e9960e1ce
commit c11484b2e7
14 changed files with 184 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-websocket/src/main/java/org/springframework/web/socket/config/MessageBrokerBeanDefinitionParser.java
View File

@@ -124,11 +124,8 @@ class MessageBrokerBeanDefinitionParser implements BeanDefinitionParser {
beanName = registerBeanDef(beanDef, parserCxt, source);
RuntimeBeanReference userSessionRegistry = new RuntimeBeanReference(beanName);
String frameBufferSizeAttribute = element.getAttribute("message-buffer-size");
Integer messageBufferSizeLimit = frameBufferSizeAttribute.isEmpty() ? null : Integer.parseInt(frameBufferSizeAttribute);
RuntimeBeanReference subProtocolWsHandler = registerSubProtocolWebSocketHandler(
clientInChannel, clientOutChannel, userSessionRegistry, messageBufferSizeLimit, parserCxt, source);
element, clientInChannel, clientOutChannel, userSessionRegistry, parserCxt, source);
for(Element stompEndpointElem : DomUtils.getChildElementsByTagName(element, "stomp-endpoint")) {
@@ -229,16 +226,12 @@ class MessageBrokerBeanDefinitionParser implements BeanDefinitionParser {
return new RuntimeBeanReference(channelName);
}
private RuntimeBeanReference registerSubProtocolWebSocketHandler(
private RuntimeBeanReference registerSubProtocolWebSocketHandler(Element element,
RuntimeBeanReference clientInChannel, RuntimeBeanReference clientOutChannel,
RuntimeBeanReference userSessionRegistry, Integer messageBufferSizeLimit,
ParserContext parserCxt, Object source) {
RuntimeBeanReference userSessionRegistry, ParserContext parserCxt, Object source) {
RootBeanDefinition stompHandlerDef = new RootBeanDefinition(StompSubProtocolHandler.class);
stompHandlerDef.getPropertyValues().add("userSessionRegistry", userSessionRegistry);
if(messageBufferSizeLimit != null) {
stompHandlerDef.getPropertyValues().add("messageBufferSizeLimit", messageBufferSizeLimit);
}
registerBeanDef(stompHandlerDef, parserCxt, source);
ConstructorArgumentValues cavs = new ConstructorArgumentValues();
@@ -248,6 +241,23 @@ class MessageBrokerBeanDefinitionParser implements BeanDefinitionParser {
RootBeanDefinition subProtocolWshDef = new RootBeanDefinition(SubProtocolWebSocketHandler.class, cavs, null);
subProtocolWshDef.getPropertyValues().addPropertyValue("protocolHandlers", stompHandlerDef);
String subProtocolWshName = registerBeanDef(subProtocolWshDef, parserCxt, source);
Element transportElem = DomUtils.getChildElementByTagName(element, "transport");
if (transportElem != null) {
String messageSize = transportElem.getAttribute("message-size");
if (messageSize != null) {
stompHandlerDef.getPropertyValues().add("messageSizeLimit", messageSize);
}
String sendTimeLimit = transportElem.getAttribute("send-timeout");
if (sendTimeLimit != null) {
subProtocolWshDef.getPropertyValues().add("sendTimeLimit", sendTimeLimit);
}
String sendBufferSizeLimit = transportElem.getAttribute("send-buffer-size");
if (sendBufferSizeLimit != null) {
subProtocolWshDef.getPropertyValues().add("sendBufferSizeLimit", sendBufferSizeLimit);
}
}
return new RuntimeBeanReference(subProtocolWshName);
}

2
spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/AbstractWebSocketMessageBrokerConfigurer.java
View File

@@ -33,7 +33,7 @@ public abstract class AbstractWebSocketMessageBrokerConfigurer implements WebSoc
@Override
public void configureWebSocketTransport(WebSocketTransportRegistration registry) {
public void configureWebSocketTransport(WebSocketTransportRegistration registration) {
}
@Override

7
spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/DelegatingWebSocketMessageBrokerConfiguration.java
View File

@@ -58,6 +58,13 @@ public class DelegatingWebSocketMessageBrokerConfiguration extends WebSocketMess
}
}
@Override
protected void configureWebSocketTransport(WebSocketTransportRegistration registration) {
for (WebSocketMessageBrokerConfigurer c : this.configurers) {
c.configureWebSocketTransport(registration);
}
}
@Override
protected void configureClientInboundChannel(ChannelRegistration registration) {
for (WebSocketMessageBrokerConfigurer c : this.configurers) {

15
spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebMvcStompEndpointRegistry.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.
@@ -21,7 +21,6 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.messaging.simp.user.UserSessionRegistry;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
@@ -58,8 +57,8 @@ public class WebMvcStompEndpointRegistry implements StompEndpointRegistry {
public WebMvcStompEndpointRegistry(WebSocketHandler webSocketHandler,
WebSocketTransportRegistration transportRegistration, UserSessionRegistry userSessionRegistry,
TaskScheduler defaultSockJsTaskScheduler, MessageBrokerRegistry brokerRegistry) {
WebSocketTransportRegistration transportRegistration,
UserSessionRegistry userSessionRegistry, TaskScheduler defaultSockJsTaskScheduler) {
Assert.notNull(webSocketHandler, "'webSocketHandler' is required ");
Assert.notNull(transportRegistration, "'transportRegistration' is required");
@@ -78,15 +77,11 @@ public class WebMvcStompEndpointRegistry implements StompEndpointRegistry {
this.stompHandler = new StompSubProtocolHandler();
this.stompHandler.setUserSessionRegistry(userSessionRegistry);
if (transportRegistration.getMessageBufferSizeLimit() != null) {
this.stompHandler.setMessageBufferSizeLimit(transportRegistration.getMessageBufferSizeLimit());
if (transportRegistration.getMessageSizeLimit() != null) {
this.stompHandler.setMessageSizeLimit(transportRegistration.getMessageSizeLimit());
}
this.sockJsScheduler = defaultSockJsTaskScheduler;
if(brokerRegistry.getMessageBufferSizeLimit() != null) {
this.stompHandler.setMessageBufferSizeLimit(brokerRegistry.getMessageBufferSizeLimit());
}
}
private static SubProtocolWebSocketHandler unwrapSubProtocolWebSocketHandler(WebSocketHandler wsHandler) {

3
spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebSocketMessageBrokerConfigurationSupport.java
View File

@@ -51,10 +51,9 @@ public abstract class WebSocketMessageBrokerConfigurationSupport extends Abstrac
UserSessionRegistry sessionRegistry = userSessionRegistry();
WebSocketTransportRegistration transportRegistration = getTransportRegistration();
ThreadPoolTaskScheduler taskScheduler = messageBrokerSockJsTaskScheduler();
MessageBrokerRegistry brokerRegistry = getBrokerRegistry();
WebMvcStompEndpointRegistry registry = new WebMvcStompEndpointRegistry(
webSocketHandler, transportRegistration, sessionRegistry, taskScheduler, brokerRegistry);
webSocketHandler, transportRegistration, sessionRegistry, taskScheduler);
registerStompEndpoints(registry);

16
spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebSocketTransportRegistration.java
View File

@@ -24,7 +24,7 @@ package org.springframework.web.socket.config.annotation;
*/
public class WebSocketTransportRegistration {
private Integer messageBufferSizeLimit;
private Integer messageSizeLimit;
private Integer sendTimeLimit;
@@ -32,9 +32,9 @@ public class WebSocketTransportRegistration {
/**
* Configure the maximum size of the buffer to use when an incoming message
* for a sub-protocol (e.g. STOMP) has been split into multiple WebSocket
* messages or multiple HTTP POSTs when SockJS fallback options are in use.
* Configure the maximum size for an incoming sub-protocol message.
* For example a STOMP message may be received as multiple WebSocket messages
* or multiple HTTP POST requests when SockJS fallback options are in use.
*
* <p>In theory a WebSocket message can be almost unlimited in size.
* In practice WebSocket servers impose limits on incoming message size.
@@ -50,16 +50,16 @@ public class WebSocketTransportRegistration {
* Version 2 of the spec will but in the mean time existing client libraries
* have already established a practice that servers must handle.
*/
public WebSocketTransportRegistration setMessageBufferSizeLimit(int bufferSizeLimit) {
this.messageBufferSizeLimit = bufferSizeLimit;
public WebSocketTransportRegistration setMessageSizeLimit(int messageSizeLimit) {
this.messageSizeLimit = messageSizeLimit;
return this;
}
/**
* Protected accessor for internal use.
*/
protected Integer getMessageBufferSizeLimit() {
return this.messageBufferSizeLimit;
protected Integer getMessageSizeLimit() {
return this.messageSizeLimit;
}
/**

24
spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java
View File

@@ -34,7 +34,6 @@ import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.simp.stomp.BufferingStompDecoder;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompConversionException;
import org.springframework.messaging.simp.stomp.StompDecoder;
import org.springframework.messaging.simp.stomp.StompEncoder;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.messaging.simp.user.DestinationUserNameProvider;
@@ -69,7 +68,7 @@ public class StompSubProtocolHandler implements SubProtocolHandler {
private static final Log logger = LogFactory.getLog(StompSubProtocolHandler.class);
private int messageBufferSizeLimit = 64 * 1024;
private int messageSizeLimit = 64 * 1024;
private final Map<String, BufferingStompDecoder> decoders = new ConcurrentHashMap<String, BufferingStompDecoder>();
@@ -79,20 +78,17 @@ public class StompSubProtocolHandler implements SubProtocolHandler {
/**
* Configure the maximum size of the buffer used when a STOMP message has been
* split over multiple WebSocket messages.
*
* <p>While the STOMP spec version 1.2 (current as of 4.0.3) does not discuss
* STOMP over WebSocket explicitly, a number of clients already split messages
* around 16K boundaries. Therefore partial content must be buffered before a
* full message can be assembled.
* Configure the maximum size allowed for an incoming STOMP message.
* Since a STOMP message can be received in multiple WebSocket messages,
* buffering may be required and therefore it is necessary to know the maximum
* allowed message size.
*
* <p>By default this property is set to 64K.
*
* @since 4.0.3
*/
public void setMessageBufferSizeLimit(int messageBufferSizeLimit) {
this.messageBufferSizeLimit = messageBufferSizeLimit;
public void setMessageSizeLimit(int messageSizeLimit) {
this.messageSizeLimit = messageSizeLimit;
}
/**
@@ -100,8 +96,8 @@ public class StompSubProtocolHandler implements SubProtocolHandler {
*
* @since 4.0.3
*/
public int getMessageBufferSizeLimit() {
return this.messageBufferSizeLimit;
public int getMessageSizeLimit() {
return this.messageSizeLimit;
}
/**
@@ -316,7 +312,7 @@ public class StompSubProtocolHandler implements SubProtocolHandler {
@Override
public void afterSessionStarted(WebSocketSession session, MessageChannel outputChannel) {
this.decoders.put(session.getId(), new BufferingStompDecoder(getMessageBufferSizeLimit()));
this.decoders.put(session.getId(), new BufferingStompDecoder(getMessageSizeLimit()));
}
@Override

112
spring-websocket/src/main/resources/org/springframework/web/socket/config/spring-websocket-4.0.xsd
View File

@@ -457,17 +457,114 @@
<xsd:annotation>
<xsd:documentation><![CDATA[
Configures broker-backed messaging over WebSocket using a higher-level messaging sub-protocol.
Registers a SimpleUrlHandlerMapping and maps
paths to registered Controllers.
Registers a SimpleUrlHandlerMapping and maps paths to registered Controllers.
A StompSubProtocolHandler is registered
to handle various versions of the STOMP protocol.
A StompSubProtocolHandler is registered to handle various versions of the STOMP protocol.
See EnableWebSocketMessageBroker javadoc for information on code-based alternatives to enabling broker-backed messaging.
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element name="transport" minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation><![CDATA[
Configure options related to the processing of messages received from and sent to WebSocket clients.
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="message-size" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Configure the maximum size for an incoming sub-protocol message.
For example a STOMP message may be received as multiple WebSocket messages
or multiple HTTP POST requests when SockJS fallback options are in use.
In theory a WebSocket message can be almost unlimited in size.
In practice WebSocket servers impose limits on incoming message size.
STOMP clients for example tend to split large messages around 16K
boundaries. Therefore a server must be able to buffer partial content
and decode when enough data is received. Use this property to configure
the max size of the buffer to use.
The default value is 64K (i.e. 64 * 1024).
NOTE that the current version 1.2 of the STOMP spec
does not specifically discuss how to send STOMP messages over WebSocket.
Version 2 of the spec will but in the mean time existing client libraries
have already established a practice that servers must handle.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="send-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Configure a time limit (in milliseconds) for the maximum amount of a time
allowed when sending messages to a WebSocket session or writing to an
HTTP response when SockJS fallback option are in use.
In general WebSocket servers expect that messages to a single WebSocket
session are sent from a single thread at a time. This is automatically
guaranteed when using {@code @EnableWebSocketMessageBroker} configuration.
If message sending is slow, or at least slower than rate of messages sending,
subsequent messages are buffered until either the {@code sendTimeLimit}
or the {@code sendBufferSizeLimit} are reached at which point the session
state is cleared and an attempt is made to close the session.
NOTE that the session time limit is checked only
on attempts to send additional messages. So if only a single message is
sent and it hangs, the session will not time out until another message is
sent or the underlying physical socket times out. So this is not a
replacement for WebSocket server or HTTP connection timeout but is rather
intended to control the extent of buffering of unsent messages.
NOTE that closing the session may not succeed in
actually closing the physical socket and may also hang. This is true
especially when using blocking IO such as the BIO connector in Tomcat
that is used by default on Tomcat 7. Therefore it is recommended to ensure
the server is using non-blocking IO such as Tomcat's NIO connector that
is used by default on Tomcat 8. If you must use blocking IO consider
customizing OS-level TCP settings, for example
{@code /proc/sys/net/ipv4/tcp_retries2} on Linux.
The default value is 10 seconds (i.e. 10 * 10000).
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="send-buffer-size" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Configure the maximum amount of data to buffer when sending messages
to a WebSocket session, or an HTTP response when SockJS fallback
option are in use.
In general WebSocket servers expect that messages to a single WebSocket
session are sent from a single thread at a time. This is automatically
guaranteed when using {@code @EnableWebSocketMessageBroker} configuration.
If message sending is slow, or at least slower than rate of messages sending,
subsequent messages are buffered until either the {@code sendTimeLimit}
or the {@code sendBufferSizeLimit} are reached at which point the session
state is cleared and an attempt is made to close the session.
NOTE that closing the session may not succeed in
actually closing the physical socket and may also hang. This is true
especially when using blocking IO such as the BIO connector in Tomcat
configured by default on Tomcat 7. Therefore it is recommended to ensure
the server is using non-blocking IO such as Tomcat's NIO connector used
by default on Tomcat 8. If you must use blocking IO consider customizing
OS-level TCP settings, for example {@code /proc/sys/net/ipv4/tcp_retries2}
on Linux.
The default value is 512K (i.e. 512 * 1024).
@param sendBufferSizeLimit the maximum number of bytes to buffer when
sending messages; if the value is less than or equal to 0 then buffering
is effectively disabled.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="stomp-endpoint" minOccurs="1" maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -575,13 +672,6 @@
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-buffer-size" type="xsd:int">
<xsd:annotation>
<xsd:documentation><![CDATA[
The message buffer size limit in bytes for simple messaging protocols like STOMP.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order" type="xsd:token">
<xsd:annotation>
<xsd:documentation><![CDATA[