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

Add WebSocket scope

Browse Source 
This change adds support for a custom "websocket" scope.

WebSocket-scoped beans may be injected into controllers with message
handling methods as well as channel interceptor registered on the
"inboundClientChannel".

Issue: SPR-11305
This commit is contained in:
Rossen Stoyanchev
2014-05-09 16:51:14 -04:00
parent 66c63c374b
commit 2c4cbb617e
18 changed files with 1090 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,8 +17,12 @@
package org.springframework.web.socket.config;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.springframework.beans.factory.config.CustomScopeConfigurer;
import org.springframework.messaging.simp.SimpSessionScope;
import org.w3c.dom.Element;
import org.springframework.beans.MutablePropertyValues;
@@ -167,6 +171,11 @@ class MessageBrokerBeanDefinitionParser implements BeanDefinitionParser {
registerUserDestinationMessageHandler(clientInChannel, clientOutChannel, brokerChannel,
userDestinationResolver, parserCxt, source);
Map<String, Object> scopeMap = Collections.<String, Object>singletonMap("websocket", new SimpSessionScope());
RootBeanDefinition scopeConfigurerDef = new RootBeanDefinition(CustomScopeConfigurer.class);
scopeConfigurerDef.getPropertyValues().add("scopes", scopeMap);
registerBeanDefByName("webSocketScopeConfigurer", scopeConfigurerDef, parserCxt, source);
parserCxt.popAndRegisterContainingComponent();
return null;

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

@@ -58,6 +58,6 @@ import org.springframework.context.annotation.Import;
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
@Documented
@Import(DelegatingWebSocketConfiguration.class)
@Import({DelegatingWebSocketConfiguration.class, WebSocketScopeConfiguration.class})
public @interface EnableWebSocket {
}

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

@@ -16,7 +16,9 @@
package org.springframework.web.socket.config.annotation;
import org.springframework.beans.factory.config.CustomScopeConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.simp.SimpSessionScope;
import org.springframework.messaging.simp.config.AbstractMessageBrokerConfiguration;
import org.springframework.messaging.simp.user.UserSessionRegistry;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -24,6 +26,8 @@ import org.springframework.web.servlet.HandlerMapping;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.messaging.SubProtocolWebSocketHandler;
import java.util.Collections;
/**
* Extends {@link AbstractMessageBrokerConfiguration} and adds configuration for
* receiving and responding to STOMP messages from WebSocket clients.
@@ -75,6 +79,8 @@ public abstract class WebSocketMessageBrokerConfigurationSupport extends Abstrac
protected void configureWebSocketTransport(WebSocketTransportRegistration registry) {
}
protected abstract void registerStompEndpoints(StompEndpointRegistry registry);
/**
* The default TaskScheduler to use if none is configured via
* {@link SockJsServiceRegistration#setTaskScheduler(org.springframework.scheduling.TaskScheduler)}, i.e.
@@ -100,6 +106,11 @@ public abstract class WebSocketMessageBrokerConfigurationSupport extends Abstrac
return scheduler;
}
protected abstract void registerStompEndpoints(StompEndpointRegistry registry);
@Bean
public static CustomScopeConfigurer webSocketScopeConfigurer() {
CustomScopeConfigurer configurer = new CustomScopeConfigurer();
configurer.setScopes(Collections.<String, Object>singletonMap("websocket", new SimpSessionScope()));
return configurer;
}
}

39
spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebSocketScopeConfiguration.java Normal file
View File

@@ -0,0 +1,39 @@
/*
* 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.web.socket.config.annotation;
import org.springframework.beans.factory.config.CustomScopeConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.simp.SimpSessionScope;
import java.util.Collections;
/**
*
* @author Rossen Stoyanchev
* @since 4.1
*/
public class WebSocketScopeConfiguration {
@Bean
public CustomScopeConfigurer webSocketScopeConfigurer() {
CustomScopeConfigurer configurer = new CustomScopeConfigurer();
configurer.setScopes(Collections.<String, Object>singletonMap("websocket", new SimpSessionScope()));
return configurer;
}
}

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

@@ -33,6 +33,8 @@ import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.simp.SimpAttributes;
import org.springframework.messaging.simp.SimpAttributesContextHolder;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.simp.stomp.BufferingStompDecoder;
@@ -221,7 +223,13 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
publishEvent(new SessionConnectEvent(this, message));
}
outputChannel.send(message);
try {
SimpAttributesContextHolder.setAttributesFromMessage(message);
outputChannel.send(message);
}
finally {
SimpAttributesContextHolder.resetAttributes();
}
}
catch (Throwable ex) {
logger.error("Terminating STOMP session due to failure to send message", ex);
@@ -420,22 +428,33 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
this.userSessionRegistry.unregisterSessionId(userName, session.getId());
}
if (logger.isDebugEnabled()) {
logger.debug("WebSocket session ended, sending DISCONNECT message to broker");
if (this.eventPublisher != null) {
publishEvent(new SessionDisconnectEvent(this, session.getId(), closeStatus));
}
Message<?> message = createDisconnectMessage(session);
SimpAttributes simpAttributes = SimpAttributes.fromMessage(message);
try {
if (logger.isDebugEnabled()) {
logger.debug("WebSocket session ended, sending DISCONNECT message to broker");
}
SimpAttributesContextHolder.setAttributes(simpAttributes);
outputChannel.send(message);
}
finally {
SimpAttributesContextHolder.resetAttributes();
simpAttributes.sessionCompleted();
}
}
private Message<?> createDisconnectMessage(WebSocketSession session) {
StompHeaderAccessor headerAccessor = StompHeaderAccessor.create(StompCommand.DISCONNECT);
if (getHeaderInitializer() != null) {
getHeaderInitializer().initHeaders(headerAccessor);
}
headerAccessor.setSessionId(session.getId());
Message<?> message = MessageBuilder.createMessage(EMPTY_PAYLOAD, headerAccessor.getMessageHeaders());
if (this.eventPublisher != null) {
publishEvent(new SessionDisconnectEvent(this, session.getId(), closeStatus));
}
outputChannel.send(message);
headerAccessor.setSessionAttributes(session.getAttributes());
return MessageBuilder.createMessage(EMPTY_PAYLOAD, headerAccessor.getMessageHeaders());
}
}

3
spring-websocket/src/test/java/org/springframework/web/socket/config/MessageBrokerBeanDefinitionParserTests.java
View File

@@ -26,6 +26,7 @@ import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.config.CustomScopeConfigurer;
import org.springframework.beans.factory.xml.XmlBeanDefinitionReader;
import org.springframework.core.io.ClassPathResource;
import org.springframework.messaging.MessageHandler;
@@ -175,6 +176,8 @@ public class MessageBrokerBeanDefinitionParserTests {
catch (NoSuchBeanDefinitionException ex) {
// expected
}
assertNotNull(this.appContext.getBean("webSocketScopeConfigurer", CustomScopeConfigurer.class));
}
@Test

38
spring-websocket/src/test/java/org/springframework/web/socket/messaging/StompSubProtocolHandlerTests.java
View File

@@ -31,6 +31,8 @@ import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.simp.SimpAttributes;
import org.springframework.messaging.simp.SimpAttributesContextHolder;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.simp.TestPrincipal;
@@ -48,6 +50,7 @@ import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.handler.TestWebSocketSession;
import org.springframework.web.socket.sockjs.transport.SockJsSession;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;
@@ -289,6 +292,41 @@ public class StompSubProtocolHandlerTests {
assertTrue(actual.getPayload().startsWith("ERROR"));
}
@Test
public void webSocketScope() {
Runnable runnable = Mockito.mock(Runnable.class);
SimpAttributes simpAttributes = new SimpAttributes(this.session.getId(), this.session.getAttributes());
simpAttributes.setAttribute("name", "value");
simpAttributes.registerDestructionCallback("name", runnable);
MessageChannel testChannel = new MessageChannel() {
@Override
public boolean send(Message<?> message) {
SimpAttributes simpAttributes = SimpAttributesContextHolder.currentAttributes();
assertThat(simpAttributes.getAttribute("name"), is("value"));
return true;
}
@Override
public boolean send(Message<?> message, long timeout) {
return false;
}
};
this.protocolHandler.afterSessionStarted(this.session, this.channel);
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.CONNECT);
Message<byte[]> message = MessageBuilder.createMessage(EMPTY_PAYLOAD, headers.getMessageHeaders());
TextMessage textMessage = new TextMessage(new StompEncoder().encode(message));
this.protocolHandler.handleMessageFromClient(this.session, textMessage, testChannel);
assertEquals(Collections.emptyList(), session.getSentMessages());
this.protocolHandler.afterSessionEnded(this.session, CloseStatus.BAD_DATA, testChannel);
assertEquals(Collections.emptyList(), session.getSentMessages());
verify(runnable, times(1)).run();
}
private static class UniqueUser extends TestPrincipal implements DestinationUserNameProvider {

92
spring-websocket/src/test/java/org/springframework/web/socket/messaging/StompWebSocketIntegrationTests.java
View File

@@ -34,6 +34,8 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.context.annotation.ScopedProxyMode;
import org.springframework.messaging.handler.annotation.MessageExceptionHandler;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.simp.annotation.SendToUser;
@@ -97,11 +99,11 @@ public class StompWebSocketIntegrationTests extends AbstractWebSocketIntegration
@Test
public void sendMessageToControllerAndReceiveReplyViaTopic() throws Exception {
TextMessage message1 = create(StompCommand.SUBSCRIBE).headers(
"id:subs1", "destination:/topic/increment").build();
TextMessage message1 = create(StompCommand.SUBSCRIBE)
.headers("id:subs1", "destination:/topic/increment").build();
TextMessage message2 = create(StompCommand.SEND).headers(
"destination:/app/increment").body("5").build();
TextMessage message2 = create(StompCommand.SEND)
.headers("destination:/app/increment").body("5").build();
TestClientWebSocketHandler clientHandler = new TestClientWebSocketHandler(1, message1, message2);
WebSocketSession session = doHandshake(clientHandler, "/ws").get();
@@ -181,6 +183,37 @@ public class StompWebSocketIntegrationTests extends AbstractWebSocketIntegration
}
}
@Test
public void webSocketScope() throws Exception {
TextMessage message1 = create(StompCommand.SUBSCRIBE)
.headers("id:subs1", "destination:/topic/scopedBeanValue").build();
TextMessage message2 = create(StompCommand.SEND)
.headers("destination:/app/scopedBeanValue").build();
TestClientWebSocketHandler clientHandler = new TestClientWebSocketHandler(1, message1, message2);
WebSocketSession session = doHandshake(clientHandler, "/ws").get();
try {
assertTrue(clientHandler.latch.await(2, TimeUnit.SECONDS));
String payload = clientHandler.actual.get(0).getPayload();
assertTrue(payload.startsWith("MESSAGE\n"));
assertTrue(payload.contains("destination:/topic/scopedBeanValue\n"));
assertTrue(payload.endsWith("\"55\"\0"));
}
finally {
session.close();
}
}
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Controller
private @interface IntegrationTestController {
}
@IntegrationTestController
static class SimpleController {
@@ -218,6 +251,42 @@ public class StompWebSocketIntegrationTests extends AbstractWebSocketIntegration
}
}
@IntegrationTestController
static class ScopedBeanController {
private final ScopedBean scopedBean;
@Autowired
public ScopedBeanController(ScopedBean scopedBean) {
this.scopedBean = scopedBean;
}
@MessageMapping(value="/scopedBeanValue")
public String getValue() {
return this.scopedBean.getValue();
}
}
static interface ScopedBean {
String getValue();
}
static class ScopedBeanImpl implements ScopedBean {
private final String value;
public ScopedBeanImpl(String value) {
this.value = value;
}
@Override
public String getValue() {
return this.value;
}
}
private static class TestClientWebSocketHandler extends TextWebSocketHandler {
@@ -251,7 +320,8 @@ public class StompWebSocketIntegrationTests extends AbstractWebSocketIntegration
}
@Configuration
@ComponentScan(basePackageClasses=StompWebSocketIntegrationTests.class,
@ComponentScan(
basePackageClasses=StompWebSocketIntegrationTests.class,
useDefaultFilters=false,
includeFilters=@ComponentScan.Filter(IntegrationTestController.class))
static class TestMessageBrokerConfigurer extends AbstractWebSocketMessageBrokerConfigurer {
@@ -269,6 +339,12 @@ public class StompWebSocketIntegrationTests extends AbstractWebSocketIntegration
configurer.setApplicationDestinationPrefixes("/app");
configurer.enableSimpleBroker("/topic", "/queue");
}
@Bean
@Scope(value="websocket", proxyMode=ScopedProxyMode.INTERFACES)
public ScopedBean scopedBean() {
return new ScopedBeanImpl("55");
}
}
@Configuration
@@ -287,10 +363,4 @@ public class StompWebSocketIntegrationTests extends AbstractWebSocketIntegration
}
}
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Controller
private @interface IntegrationTestController {
}
}