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

Consolidate websocket/messaging code

Browse Source 
Before this change spring-messaging contained a few WebSocket-related
classes including WebSocket sub-protocol support for STOMP as well
as @EnableWebSocketMessageBroker and related configuration classes.

After this change those classes are located in the spring-websocket
module under org.springframework.web.socket.messaging.

This means the following classes in application configuration must
have their packages updated:

org.springframework.web.socket.messaging.config.EnableWebSocketMessageBroker
org.springframework.web.socket.messaging.config.StompEndpointRegistry
org.springframework.web.socket.messaging.config.WebSocketMessageBrokerConfigurer

MessageBrokerConfigurer has been renamed to MessageBrokerRegistry and
is also located in the above package.
This commit is contained in:
Rossen Stoyanchev
2013-11-22 13:49:38 -05:00
parent f888b8816f
commit 4de3291dc7
34 changed files with 966 additions and 870 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
spring-messaging/src/main/java/org/springframework/messaging/handler/websocket/SubProtocolHandler.java
View File

@@ -1,94 +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.messaging.handler.websocket;
import java.util.List;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;
/**
* A contract for handling WebSocket messages as part of a higher level protocol, referred
* to as "sub-protocol" in the WebSocket RFC specification. Handles both
* {@link WebSocketMessage}s from a client as well as {@link Message}s to a client.
* <p>
* Implementations of this interface can be configured on a
* {@link SubProtocolWebSocketHandler} which selects a sub-protocol handler to delegate
* messages to based on the sub-protocol requested by the client through the
* {@code Sec-WebSocket-Protocol} request header.
*
* @author Andy Wilkinson
* @author Rossen Stoyanchev
*
* @since 4.0
*/
public interface SubProtocolHandler {
/**
* Return the list of sub-protocols supported by this handler, never {@code null}.
*/
List<String> getSupportedProtocols();
/**
* Handle the given {@link WebSocketMessage} received from a client.
*
* @param session the client session
* @param message the client message
* @param outputChannel an output channel to send messages to
*/
void handleMessageFromClient(WebSocketSession session, WebSocketMessage message,
MessageChannel outputChannel) throws Exception;
/**
* Handle the given {@link Message} to the client associated with the given WebSocket
* session.
*
* @param session the client session
* @param message the client message
*/
void handleMessageToClient(WebSocketSession session, Message<?> message) throws Exception;
/**
* Resolve the session id from the given message or return {@code null}.
*
* @param message the message to resolve the session id from
*/
String resolveSessionId(Message<?> message);
/**
* Invoked after a {@link WebSocketSession} has started.
*
* @param session the client session
* @param outputChannel a channel
*/
void afterSessionStarted(WebSocketSession session, MessageChannel outputChannel) throws Exception;
/**
* Invoked after a {@link WebSocketSession} has ended.
*
* @param session the client session
* @param closeStatus the reason why the session was closed
* @param outputChannel a channel
*/
void afterSessionEnded(WebSocketSession session, CloseStatus closeStatus,
MessageChannel outputChannel) throws Exception;
}

232
spring-messaging/src/main/java/org/springframework/messaging/handler/websocket/SubProtocolWebSocketHandler.java
View File

@@ -1,232 +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.messaging.handler.websocket;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;
/**
* A {@link WebSocketHandler} that delegates messages to a {@link SubProtocolHandler}
* based on the sub-protocol value requested by the client through the
* {@code Sec-WebSocket-Protocol} request header A default handler can also be configured
* to use if the client does not request a specific sub-protocol.
*
* @author Rossen Stoyanchev
* @author Andy Wilkinson
*
* @since 4.0
*/
public class SubProtocolWebSocketHandler implements WebSocketHandler, MessageHandler {
private final Log logger = LogFactory.getLog(SubProtocolWebSocketHandler.class);
private final MessageChannel outputChannel;
private final Map<String, SubProtocolHandler> protocolHandlers =
new TreeMap<String, SubProtocolHandler>(String.CASE_INSENSITIVE_ORDER);
private SubProtocolHandler defaultProtocolHandler;
private final Map<String, WebSocketSession> sessions = new ConcurrentHashMap<String, WebSocketSession>();
/**
* @param outputChannel
*/
public SubProtocolWebSocketHandler(MessageChannel outputChannel) {
Assert.notNull(outputChannel, "outputChannel is required");
this.outputChannel = outputChannel;
}
/**
* Configure one or more handlers to use depending on the sub-protocol requested by
* the client in the WebSocket handshake request.
*
* @param protocolHandlers the sub-protocol handlers to use
*/
public void setProtocolHandlers(List<SubProtocolHandler> protocolHandlers) {
this.protocolHandlers.clear();
for (SubProtocolHandler handler: protocolHandlers) {
addProtocolHandler(handler);
}
}
/**
* Register a sub-protocol handler.
*/
public void addProtocolHandler(SubProtocolHandler handler) {
List<String> protocols = handler.getSupportedProtocols();
if (CollectionUtils.isEmpty(protocols)) {
logger.warn("No sub-protocols, ignoring handler " + handler);
return;
}
for (String protocol: protocols) {
SubProtocolHandler replaced = this.protocolHandlers.put(protocol, handler);
if ((replaced != null) && (replaced != handler) ) {
throw new IllegalStateException("Failed to map handler " + handler
+ " to protocol '" + protocol + "', it is already mapped to handler " + replaced);
}
}
}
/**
* @return the configured sub-protocol handlers
*/
public Map<String, SubProtocolHandler> getProtocolHandlers() {
return this.protocolHandlers;
}
/**
* Set the {@link SubProtocolHandler} to use when the client did not request a
* sub-protocol.
*
* @param defaultProtocolHandler the default handler
*/
public void setDefaultProtocolHandler(SubProtocolHandler defaultProtocolHandler) {
this.defaultProtocolHandler = defaultProtocolHandler;
if (this.protocolHandlers.isEmpty()) {
setProtocolHandlers(Arrays.asList(defaultProtocolHandler));
}
}
/**
* @return the default sub-protocol handler to use
*/
public SubProtocolHandler getDefaultProtocolHandler() {
return this.defaultProtocolHandler;
}
/**
* Return all supported protocols.
*/
public Set<String> getSupportedProtocols() {
return this.protocolHandlers.keySet();
}
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
this.sessions.put(session.getId(), session);
findProtocolHandler(session).afterSessionStarted(session, this.outputChannel);
}
protected final SubProtocolHandler findProtocolHandler(WebSocketSession session) {
SubProtocolHandler handler;
String protocol = session.getAcceptedProtocol();
if (!StringUtils.isEmpty(protocol)) {
handler = this.protocolHandlers.get(protocol);
Assert.state(handler != null,
"No handler for sub-protocol '" + protocol + "', handlers=" + this.protocolHandlers);
}
else {
if (this.defaultProtocolHandler != null) {
handler = this.defaultProtocolHandler;
}
else {
Set<SubProtocolHandler> handlers = new HashSet<SubProtocolHandler>(this.protocolHandlers.values());
if (handlers.size() == 1) {
handler = handlers.iterator().next();
}
else {
throw new IllegalStateException(
"No sub-protocol was requested and a default sub-protocol handler was not configured");
}
}
}
return handler;
}
@Override
public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
findProtocolHandler(session).handleMessageFromClient(session, message, this.outputChannel);
}
@Override
public void handleMessage(Message<?> message) throws MessagingException {
String sessionId = resolveSessionId(message);
if (sessionId == null) {
logger.error("sessionId not found in message " + message);
return;
}
WebSocketSession session = this.sessions.get(sessionId);
if (session == null) {
logger.error("Session not found for session with id " + sessionId);
return;
}
try {
findProtocolHandler(session).handleMessageToClient(session, message);
}
catch (Exception e) {
logger.error("Failed to send message to client " + message, e);
}
}
private String resolveSessionId(Message<?> message) {
for (SubProtocolHandler handler : this.protocolHandlers.values()) {
String sessionId = handler.resolveSessionId(message);
if (sessionId != null) {
return sessionId;
}
}
if (this.defaultProtocolHandler != null) {
String sessionId = this.defaultProtocolHandler.resolveSessionId(message);
if (sessionId != null) {
return sessionId;
}
}
return null;
}
@Override
public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
}
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {
this.sessions.remove(session.getId());
findProtocolHandler(session).afterSessionEnded(session, closeStatus, this.outputChannel);
}
@Override
public boolean supportsPartialMessages() {
return false;
}
}

4
spring-messaging/src/main/java/org/springframework/messaging/simp/annotation/support/SubscriptionMethodReturnValueHandler.java
View File

@@ -47,6 +47,10 @@ public class SubscriptionMethodReturnValueHandler implements HandlerMethodReturn
private final MessageSendingOperations<String> messagingTemplate;
/**
* @param messagingTemplate a messaging template for sending messages directly
* to clients, e.g. in response to a subscription
*/
public SubscriptionMethodReturnValueHandler(MessageSendingOperations<String> messagingTemplate) {
Assert.notNull(messagingTemplate, "messagingTemplate is required");
this.messagingTemplate = messagingTemplate;

21
spring-messaging/src/main/java/org/springframework/messaging/simp/config/AbstractBrokerRegistration.java
View File

@@ -19,6 +19,7 @@ package org.springframework.messaging.simp.config;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.simp.handler.AbstractBrokerMessageHandler;
@@ -32,25 +33,25 @@ import org.springframework.util.Assert;
*/
public abstract class AbstractBrokerRegistration {
private final MessageChannel webSocketReplyChannel;
private final MessageChannel clientOutboundChannel;
private final String[] destinationPrefixes;
private final List<String> destinationPrefixes;
public AbstractBrokerRegistration(MessageChannel webSocketReplyChannel, String[] destinationPrefixes) {
Assert.notNull(webSocketReplyChannel, "");
this.webSocketReplyChannel = webSocketReplyChannel;
this.destinationPrefixes = destinationPrefixes;
public AbstractBrokerRegistration(MessageChannel clientOutboundChannel, String[] destinationPrefixes) {
Assert.notNull(clientOutboundChannel, "'clientOutboundChannel' is required");
this.clientOutboundChannel = clientOutboundChannel;
this.destinationPrefixes = (destinationPrefixes != null)
? Arrays.<String>asList(destinationPrefixes) : Collections.<String>emptyList();
}
protected MessageChannel getWebSocketReplyChannel() {
return this.webSocketReplyChannel;
protected MessageChannel getClientOutboundChannel() {
return this.clientOutboundChannel;
}
protected Collection<String> getDestinationPrefixes() {
return (this.destinationPrefixes != null)
? Arrays.<String>asList(this.destinationPrefixes) : Collections.<String>emptyList();
return this.destinationPrefixes;
}
protected abstract AbstractBrokerMessageHandler getMessageHandler();

223
spring-messaging/src/main/java/org/springframework/messaging/simp/config/AbstractMessageBrokerConfiguration.java Normal file
View File

@@ -0,0 +1,223 @@
/*
* 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.messaging.simp.config;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.handler.*;
import org.springframework.messaging.support.channel.AbstractSubscribableChannel;
import org.springframework.messaging.support.channel.ExecutorSubscribableChannel;
import org.springframework.messaging.support.converter.*;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.ClassUtils;
import org.springframework.util.MimeTypeUtils;
import java.util.ArrayList;
import java.util.List;
/**
* Provides essential configuration for handling messages with simple messaging
* protocols such as STOMP.
* <p>
* {@link #clientInboundChannel()} and {@link #clientOutboundChannel()} deliver messages
* to and from remote clients to several message handlers such as
* <ul>
* <li>{@link #simpAnnotationMethodMessageHandler()}</li>
* <li>{@link #simpleBrokerMessageHandler()}</li>
* <li>{@link #stompBrokerRelayMessageHandler()}</li>
* <li>{@link #userDestinationMessageHandler()}</li>
* </ul>
* while {@link #brokerChannel()} delivers messages from within the application to the
* the respective message handlers. {@link #brokerMessagingTemplate()} can be injected
* into any application component to send messages.
* <p>
* Sub-classes are responsible for the part of the configuration that feed messages
* to and from the client inbound/outbound channels (e.g. STOMP over WebSokcet).
*
* @author Rossen Stoyanchev
* @since 4.0
*/
public abstract class AbstractMessageBrokerConfiguration {
private static final boolean jackson2Present= ClassUtils.isPresent(
"com.fasterxml.jackson.databind.ObjectMapper", AbstractMessageBrokerConfiguration.class.getClassLoader());
private MessageBrokerRegistry brokerRegistry;
/**
* Protected constructor.
*/
protected AbstractMessageBrokerConfiguration() {
}
/**
* An accessor for the {@link MessageBrokerRegistry} that ensures its one-time creation
* and initialization through {@link #configureMessageBroker(MessageBrokerRegistry)}.
*/
protected final MessageBrokerRegistry getBrokerRegistry() {
if (this.brokerRegistry == null) {
MessageBrokerRegistry registry = new MessageBrokerRegistry(clientOutboundChannel());
configureMessageBroker(registry);
this.brokerRegistry = registry;
}
return this.brokerRegistry;
}
/**
* A hook for sub-classes to customize message broker configuration through the
* provided {@link MessageBrokerRegistry} instance.
*/
protected abstract void configureMessageBroker(MessageBrokerRegistry registry);
@Bean
public AbstractSubscribableChannel clientInboundChannel() {
return new ExecutorSubscribableChannel(clientInboundChannelExecutor());
}
@Bean
public ThreadPoolTaskExecutor clientInboundChannelExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setThreadNamePrefix("ClientInboundChannel-");
return executor;
}
@Bean
public AbstractSubscribableChannel clientOutboundChannel() {
return new ExecutorSubscribableChannel(clientOutboundChannelExecutor());
}
@Bean
public ThreadPoolTaskExecutor clientOutboundChannelExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setThreadNamePrefix("ClientOutboundChannel-");
return executor;
}
@Bean
public AbstractSubscribableChannel brokerChannel() {
return new ExecutorSubscribableChannel(); // synchronous
}
@Bean
public SimpAnnotationMethodMessageHandler simpAnnotationMethodMessageHandler() {
SimpAnnotationMethodMessageHandler handler =
new SimpAnnotationMethodMessageHandler(brokerMessagingTemplate(), clientOutboundChannel());
handler.setDestinationPrefixes(getBrokerRegistry().getApplicationDestinationPrefixes());
handler.setMessageConverter(brokerMessageConverter());
clientInboundChannel().subscribe(handler);
return handler;
}
@Bean
public AbstractBrokerMessageHandler simpleBrokerMessageHandler() {
SimpleBrokerMessageHandler handler = getBrokerRegistry().getSimpleBroker();
if (handler != null) {
clientInboundChannel().subscribe(handler);
brokerChannel().subscribe(handler);
return handler;
}
return noopBroker;
}
@Bean
public AbstractBrokerMessageHandler stompBrokerRelayMessageHandler() {
AbstractBrokerMessageHandler handler = getBrokerRegistry().getStompBrokerRelay();
if (handler != null) {
clientInboundChannel().subscribe(handler);
brokerChannel().subscribe(handler);
return handler;
}
return noopBroker;
}
@Bean
public UserDestinationMessageHandler userDestinationMessageHandler() {
UserDestinationMessageHandler handler = new UserDestinationMessageHandler(
brokerMessagingTemplate(), userDestinationResolver());
clientInboundChannel().subscribe(handler);
brokerChannel().subscribe(handler);
return handler;
}
@Bean
public SimpMessagingTemplate brokerMessagingTemplate() {
SimpMessagingTemplate template = new SimpMessagingTemplate(brokerChannel());
String prefix = getBrokerRegistry().getUserDestinationPrefix();
if (prefix != null) {
template.setUserDestinationPrefix(prefix);
}
template.setMessageConverter(brokerMessageConverter());
return template;
}
@Bean
public CompositeMessageConverter brokerMessageConverter() {
DefaultContentTypeResolver contentTypeResolver = new DefaultContentTypeResolver();
List<MessageConverter> converters = new ArrayList<MessageConverter>();
if (jackson2Present) {
converters.add(new MappingJackson2MessageConverter());
contentTypeResolver.setDefaultMimeType(MimeTypeUtils.APPLICATION_JSON);
}
converters.add(new StringMessageConverter());
converters.add(new ByteArrayMessageConverter());
return new CompositeMessageConverter(converters, contentTypeResolver);
}
@Bean
public UserDestinationResolver userDestinationResolver() {
DefaultUserDestinationResolver resolver = new DefaultUserDestinationResolver(userSessionRegistry());
String prefix = getBrokerRegistry().getUserDestinationPrefix();
if (prefix != null) {
resolver.setUserDestinationPrefix(prefix);
}
return resolver;
}
@Bean
public UserSessionRegistry userSessionRegistry() {
return new DefaultUserSessionRegistry();
}
private static final AbstractBrokerMessageHandler noopBroker = new AbstractBrokerMessageHandler(null) {
@Override
protected void startInternal() {
}
@Override
protected void stopInternal() {
}
@Override
protected void handleMessageInternal(Message<?> message) {
}
};
}

154
spring-messaging/src/main/java/org/springframework/messaging/simp/config/AbstractStompEndpointRegistration.java
View File

@@ -1,154 +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.messaging.simp.config;
import java.util.Set;
import org.springframework.messaging.handler.websocket.SubProtocolWebSocketHandler;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.DefaultHandshakeHandler;
import org.springframework.web.socket.server.HandshakeHandler;
import org.springframework.web.socket.server.config.SockJsServiceRegistration;
import org.springframework.web.socket.sockjs.SockJsService;
import org.springframework.web.socket.sockjs.transport.handler.WebSocketTransportHandler;
import org.springframework.web.socket.support.WebSocketHandlerDecorator;
/**
* An abstract base class class for configuring STOMP over WebSocket/SockJS endpoints.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
public abstract class AbstractStompEndpointRegistration<M> implements StompEndpointRegistration {
private final String[] paths;
private final WebSocketHandler wsHandler;
private HandshakeHandler handshakeHandler;
private StompSockJsServiceRegistration sockJsServiceRegistration;
private final TaskScheduler sockJsTaskScheduler;
public AbstractStompEndpointRegistration(String[] paths, WebSocketHandler webSocketHandler,
TaskScheduler sockJsTaskScheduler) {
Assert.notEmpty(paths, "No paths specified");
this.paths = paths;
this.wsHandler = webSocketHandler;
this.sockJsTaskScheduler = sockJsTaskScheduler;
}
/**
* Provide a custom or pre-configured {@link HandshakeHandler}. This property is
* optional.
*/
@Override
public StompEndpointRegistration setHandshakeHandler(HandshakeHandler handshakeHandler) {
this.handshakeHandler = handshakeHandler;
return this;
}
/**
* Enable SockJS fallback options.
*/
@Override
public SockJsServiceRegistration withSockJS() {
this.sockJsServiceRegistration = new StompSockJsServiceRegistration(this.sockJsTaskScheduler);
if (this.handshakeHandler != null) {
WebSocketTransportHandler transportHandler = new WebSocketTransportHandler(this.handshakeHandler);
this.sockJsServiceRegistration.setTransportHandlerOverrides(transportHandler);
}
return this.sockJsServiceRegistration;
}
protected final M getMappings() {
M mappings = createMappings();
if (this.sockJsServiceRegistration != null) {
SockJsService sockJsService = this.sockJsServiceRegistration.getSockJsService();
for (String path : this.paths) {
String pathPattern = path.endsWith("/") ? path + "**" : path + "/**";
addSockJsServiceMapping(mappings, sockJsService, this.wsHandler, pathPattern);
}
}
else {
HandshakeHandler handshakeHandler = getOrCreateHandshakeHandler();
for (String path : this.paths) {
addWebSocketHandlerMapping(mappings, this.wsHandler, handshakeHandler, path);
}
}
return mappings;
}
protected abstract M createMappings();
private HandshakeHandler getOrCreateHandshakeHandler() {
HandshakeHandler handler = (this.handshakeHandler != null)
? this.handshakeHandler : new DefaultHandshakeHandler();
if (handler instanceof DefaultHandshakeHandler) {
DefaultHandshakeHandler defaultHandshakeHandler = (DefaultHandshakeHandler) handler;
if (ObjectUtils.isEmpty(defaultHandshakeHandler.getSupportedProtocols())) {
Set<String> protocols = findSubProtocolWebSocketHandler(this.wsHandler).getSupportedProtocols();
defaultHandshakeHandler.setSupportedProtocols(protocols.toArray(new String[protocols.size()]));
}
}
return handler;
}
private static SubProtocolWebSocketHandler findSubProtocolWebSocketHandler(WebSocketHandler webSocketHandler) {
WebSocketHandler actual = (webSocketHandler instanceof WebSocketHandlerDecorator) ?
((WebSocketHandlerDecorator) webSocketHandler).getLastHandler() : webSocketHandler;
Assert.isInstanceOf(SubProtocolWebSocketHandler.class, actual,
"No SubProtocolWebSocketHandler found: " + webSocketHandler);
return (SubProtocolWebSocketHandler) actual;
}
protected abstract void addSockJsServiceMapping(M mappings, SockJsService sockJsService,
WebSocketHandler wsHandler, String pathPattern);
protected abstract void addWebSocketHandlerMapping(M mappings,
WebSocketHandler wsHandler, HandshakeHandler handshakeHandler, String path);
private class StompSockJsServiceRegistration extends SockJsServiceRegistration {
public StompSockJsServiceRegistration(TaskScheduler defaultTaskScheduler) {
super(defaultTaskScheduler);
}
protected SockJsService getSockJsService() {
return super.getSockJsService();
}
}
}

66
spring-messaging/src/main/java/org/springframework/messaging/simp/config/DelegatingWebSocketMessageBrokerConfiguration.java
View File

@@ -1,66 +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.messaging.simp.config;
import java.util.ArrayList;
import java.util.List;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.CollectionUtils;
/**
* A {@link WebSocketMessageBrokerConfiguration} extension that detects beans of type
* {@link WebSocketMessageBrokerConfigurer} and delegates to all of them allowing callback
* style customization of the configuration provided in
* {@link WebSocketMessageBrokerConfigurationSupport}.
*
* <p>This class is typically imported via {@link EnableWebSocketMessageBroker}.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
@Configuration
public class DelegatingWebSocketMessageBrokerConfiguration extends WebSocketMessageBrokerConfigurationSupport {
private List<WebSocketMessageBrokerConfigurer> configurers = new ArrayList<WebSocketMessageBrokerConfigurer>();
@Autowired(required=false)
public void setConfigurers(List<WebSocketMessageBrokerConfigurer> configurers) {
if (CollectionUtils.isEmpty(configurers)) {
return;
}
this.configurers.addAll(configurers);
}
@Override
protected void registerStompEndpoints(StompEndpointRegistry registry) {
for (WebSocketMessageBrokerConfigurer c : this.configurers) {
c.registerStompEndpoints(registry);
}
}
@Override
protected void configureMessageBroker(MessageBrokerConfigurer configurer) {
for (WebSocketMessageBrokerConfigurer c : this.configurers) {
c.configureMessageBroker(configurer);
}
}
}

65
spring-messaging/src/main/java/org/springframework/messaging/simp/config/EnableWebSocketMessageBroker.java
View File

@@ -1,65 +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.messaging.simp.config;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.context.annotation.Import;
/**
* Add this annotation to an {@code @Configuration} class to enable broker-backed
* messaging over WebSocket using a higher-level messaging sub-protocol.
*
* <pre class="code">
* &#064;Configuration
* &#064;EnableWebSocketMessageBroker
* public class MyWebSocketConfig {
*
* }
* </pre>
* <p>
* Customize the imported configuration by implementing the
* {@link WebSocketMessageBrokerConfigurer} interface:
*
* <pre class="code">
* &#064;Configuration
* &#064;EnableWebSocketMessageBroker
* public class MyConfiguration implements implements WebSocketMessageBrokerConfigurer {
*
* &#064;Override
* public void registerStompEndpoints(StompEndpointRegistry registry) {
* registry.addEndpoint("/portfolio").withSockJS();
* }
*
* &#064;Bean
* public void configureMessageBroker(MessageBrokerConfigurer configurer) {
* configurer.enableStompBrokerRelay("/queue/", "/topic/");
* configurer.setApplicationDestinationPrefixes("/app/");
* }
* }
* </pre>
*
* @author Rossen Stoyanchev
* @since 4.0
*/
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
@Documented
@Import(DelegatingWebSocketMessageBrokerConfiguration.class)
public @interface EnableWebSocketMessageBroker {
}

29
spring-messaging/src/main/java/org/springframework/messaging/simp/config/MessageBrokerConfigurer.java → spring-messaging/src/main/java/org/springframework/messaging/simp/config/MessageBrokerRegistry.java
View File

@@ -20,18 +20,19 @@ import java.util.Arrays;
import java.util.Collection;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.simp.handler.AbstractBrokerMessageHandler;
import org.springframework.messaging.simp.handler.SimpleBrokerMessageHandler;
import org.springframework.messaging.simp.stomp.StompBrokerRelayMessageHandler;
import org.springframework.util.Assert;
/**
* A helper class for configuring message broker options.
* A registry for configuring message broker options.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
public class MessageBrokerConfigurer {
public class MessageBrokerRegistry {
private final MessageChannel webSocketResponseChannel;
private final MessageChannel clientOutboundChannel;
private SimpleBrokerRegistration simpleBroker;
@@ -42,9 +43,9 @@ public class MessageBrokerConfigurer {
private String userDestinationPrefix;
public MessageBrokerConfigurer(MessageChannel webSocketResponseChannel) {
Assert.notNull(webSocketResponseChannel);
this.webSocketResponseChannel = webSocketResponseChannel;
public MessageBrokerRegistry(MessageChannel clientOutboundChannel) {
Assert.notNull(clientOutboundChannel);
this.clientOutboundChannel = clientOutboundChannel;
}
/**
@@ -52,7 +53,7 @@ public class MessageBrokerConfigurer {
* destinations targeting the broker (e.g. destinations prefixed with "/topic").
*/
public SimpleBrokerRegistration enableSimpleBroker(String... destinationPrefixes) {
this.simpleBroker = new SimpleBrokerRegistration(this.webSocketResponseChannel, destinationPrefixes);
this.simpleBroker = new SimpleBrokerRegistration(this.clientOutboundChannel, destinationPrefixes);
return this.simpleBroker;
}
@@ -62,7 +63,7 @@ public class MessageBrokerConfigurer {
* destinations.
*/
public StompBrokerRelayRegistration enableStompBrokerRelay(String... destinationPrefixes) {
this.stompRelay = new StompBrokerRelayRegistration(this.webSocketResponseChannel, destinationPrefixes);
this.stompRelay = new StompBrokerRelayRegistration(this.clientOutboundChannel, destinationPrefixes);
return this.stompRelay;
}
@@ -78,7 +79,7 @@ public class MessageBrokerConfigurer {
* <p>
* Prefixes that do not have a trailing slash will have one automatically appended.
*/
public MessageBrokerConfigurer setApplicationDestinationPrefixes(String... prefixes) {
public MessageBrokerRegistry setApplicationDestinationPrefixes(String... prefixes) {
this.applicationDestinationPrefixes = prefixes;
return this;
}
@@ -97,24 +98,24 @@ public class MessageBrokerConfigurer {
* <p>
* The default prefix used to identify such destinations is "/user/".
*/
public MessageBrokerConfigurer setUserDestinationPrefix(String destinationPrefix) {
public MessageBrokerRegistry setUserDestinationPrefix(String destinationPrefix) {
this.userDestinationPrefix = destinationPrefix;
return this;
}
protected AbstractBrokerMessageHandler getSimpleBroker() {
protected SimpleBrokerMessageHandler getSimpleBroker() {
initSimpleBrokerIfNecessary();
return (this.simpleBroker != null) ? this.simpleBroker.getMessageHandler() : null;
}
protected void initSimpleBrokerIfNecessary() {
if ((this.simpleBroker == null) && (this.stompRelay == null)) {
this.simpleBroker = new SimpleBrokerRegistration(this.webSocketResponseChannel, null);
this.simpleBroker = new SimpleBrokerRegistration(this.clientOutboundChannel, null);
}
}
protected AbstractBrokerMessageHandler getStompBrokerRelay() {
protected StompBrokerRelayMessageHandler getStompBrokerRelay() {
return (this.stompRelay != null) ? this.stompRelay.getMessageHandler() : null;
}

68
spring-messaging/src/main/java/org/springframework/messaging/simp/config/ServletStompEndpointRegistration.java
View File

@@ -1,68 +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.messaging.simp.config;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.HttpRequestHandler;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.HandshakeHandler;
import org.springframework.web.socket.server.support.WebSocketHttpRequestHandler;
import org.springframework.web.socket.sockjs.SockJsHttpRequestHandler;
import org.springframework.web.socket.sockjs.SockJsService;
/**
* A helper class for configuring STOMP protocol handling over WebSocket
* with optional SockJS fallback options.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
public class ServletStompEndpointRegistration
extends AbstractStompEndpointRegistration<MultiValueMap<HttpRequestHandler, String>> {
public ServletStompEndpointRegistration(String[] paths,
WebSocketHandler wsHandler, TaskScheduler sockJsTaskScheduler) {
super(paths, wsHandler, sockJsTaskScheduler);
}
@Override
protected MultiValueMap<HttpRequestHandler, String> createMappings() {
return new LinkedMultiValueMap<HttpRequestHandler, String>();
}
@Override
protected void addSockJsServiceMapping(MultiValueMap<HttpRequestHandler, String> mappings,
SockJsService sockJsService, WebSocketHandler wsHandler, String pathPattern) {
SockJsHttpRequestHandler httpHandler = new SockJsHttpRequestHandler(sockJsService, wsHandler);
mappings.add(httpHandler, pathPattern);
}
@Override
protected void addWebSocketHandlerMapping(MultiValueMap<HttpRequestHandler, String> mappings,
WebSocketHandler wsHandler, HandshakeHandler handshakeHandler, String path) {
WebSocketHttpRequestHandler handler = new WebSocketHttpRequestHandler(wsHandler, handshakeHandler);
mappings.add(handler, path);
}
}

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

@@ -1,108 +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.messaging.simp.config;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.springframework.messaging.handler.websocket.SubProtocolWebSocketHandler;
import org.springframework.messaging.simp.handler.UserSessionRegistry;
import org.springframework.messaging.simp.stomp.StompProtocolHandler;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
import org.springframework.util.MultiValueMap;
import org.springframework.web.HttpRequestHandler;
import org.springframework.web.servlet.handler.AbstractHandlerMapping;
import org.springframework.web.servlet.handler.SimpleUrlHandlerMapping;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.support.WebSocketHandlerDecorator;
/**
* A helper class for configuring STOMP protocol handling over WebSocket.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
public class ServletStompEndpointRegistry implements StompEndpointRegistry {
private final WebSocketHandler webSocketHandler;
private final SubProtocolWebSocketHandler subProtocolWebSocketHandler;
private final StompProtocolHandler stompHandler;
private final List<ServletStompEndpointRegistration> registrations = new ArrayList<ServletStompEndpointRegistration>();
private final TaskScheduler sockJsScheduler;
public ServletStompEndpointRegistry(WebSocketHandler webSocketHandler,
UserSessionRegistry userSessionRegistry, TaskScheduler defaultSockJsTaskScheduler) {
Assert.notNull(webSocketHandler);
Assert.notNull(userSessionRegistry);
this.webSocketHandler = webSocketHandler;
this.subProtocolWebSocketHandler = findSubProtocolWebSocketHandler(webSocketHandler);
this.stompHandler = new StompProtocolHandler();
this.stompHandler.setUserSessionRegistry(userSessionRegistry);
this.sockJsScheduler = defaultSockJsTaskScheduler;
}
private static SubProtocolWebSocketHandler findSubProtocolWebSocketHandler(WebSocketHandler webSocketHandler) {
WebSocketHandler actual = (webSocketHandler instanceof WebSocketHandlerDecorator) ?
((WebSocketHandlerDecorator) webSocketHandler).getLastHandler() : webSocketHandler;
Assert.isInstanceOf(SubProtocolWebSocketHandler.class, actual,
"No SubProtocolWebSocketHandler found: " + webSocketHandler);
return (SubProtocolWebSocketHandler) actual;
}
@Override
public StompEndpointRegistration addEndpoint(String... paths) {
this.subProtocolWebSocketHandler.addProtocolHandler(this.stompHandler);
ServletStompEndpointRegistration r = new ServletStompEndpointRegistration(
paths, this.webSocketHandler, this.sockJsScheduler);
this.registrations.add(r);
return r;
}
/**
* Returns a handler mapping with the mapped ViewControllers; or {@code null} in case of no registrations.
*/
protected AbstractHandlerMapping getHandlerMapping() {
Map<String, Object> urlMap = new LinkedHashMap<String, Object>();
for (ServletStompEndpointRegistration registration : this.registrations) {
MultiValueMap<HttpRequestHandler, String> mappings = registration.getMappings();
for (HttpRequestHandler httpHandler : mappings.keySet()) {
for (String pattern : mappings.get(httpHandler)) {
urlMap.put(pattern, httpHandler);
}
}
}
SimpleUrlHandlerMapping hm = new SimpleUrlHandlerMapping();
hm.setUrlMap(urlMap);
return hm;
}
}

10
spring-messaging/src/main/java/org/springframework/messaging/simp/config/SimpleBrokerRegistration.java
View File

@@ -21,7 +21,7 @@ import org.springframework.messaging.simp.handler.SimpleBrokerMessageHandler;
/**
* A simple message broker alternative providing a simple getting started option.
* Registration class for configuring a {@link SimpleBrokerMessageHandler}.
*
* @author Rossen Stoyanchev
* @since 4.0
@@ -29,15 +29,13 @@ import org.springframework.messaging.simp.handler.SimpleBrokerMessageHandler;
public class SimpleBrokerRegistration extends AbstractBrokerRegistration {
public SimpleBrokerRegistration(MessageChannel webSocketReplyChannel, String[] destinationPrefixes) {
super(webSocketReplyChannel, destinationPrefixes);
public SimpleBrokerRegistration(MessageChannel clientOutboundChannel, String[] destinationPrefixes) {
super(clientOutboundChannel, destinationPrefixes);
}
@Override
protected SimpleBrokerMessageHandler getMessageHandler() {
SimpleBrokerMessageHandler handler =
new SimpleBrokerMessageHandler(getWebSocketReplyChannel(), getDestinationPrefixes());
return handler;
return new SimpleBrokerMessageHandler(getClientOutboundChannel(), getDestinationPrefixes());
}
}

13
spring-messaging/src/main/java/org/springframework/messaging/simp/config/StompBrokerRelayRegistration.java
View File

@@ -22,7 +22,7 @@ import org.springframework.util.Assert;
/**
* A helper class for configuring a relay to an external STOMP message broker.
* Registration class for configuring a {@link StompBrokerRelayMessageHandler}.
*
* @author Rossen Stoyanchev
* @since 4.0
@@ -44,8 +44,8 @@ public class StompBrokerRelayRegistration extends AbstractBrokerRegistration {
private boolean autoStartup = true;
public StompBrokerRelayRegistration(MessageChannel webSocketReplyChannel, String[] destinationPrefixes) {
super(webSocketReplyChannel, destinationPrefixes);
public StompBrokerRelayRegistration(MessageChannel clientOutboundChannel, String[] destinationPrefixes) {
super(clientOutboundChannel, destinationPrefixes);
}
@@ -124,18 +124,23 @@ public class StompBrokerRelayRegistration extends AbstractBrokerRegistration {
protected StompBrokerRelayMessageHandler getMessageHandler() {
StompBrokerRelayMessageHandler handler =
new StompBrokerRelayMessageHandler(getWebSocketReplyChannel(), getDestinationPrefixes());
new StompBrokerRelayMessageHandler(getClientOutboundChannel(), getDestinationPrefixes());
handler.setRelayHost(this.relayHost);
handler.setRelayPort(this.relayPort);
handler.setSystemLogin(this.applicationLogin);
handler.setSystemPasscode(this.applicationPasscode);
if (this.systemHeartbeatSendInterval != null) {
handler.setSystemHeartbeatSendInterval(this.systemHeartbeatSendInterval);
}
if (this.systemHeartbeatReceiveInterval != null) {
handler.setSystemHeartbeatReceiveInterval(this.systemHeartbeatReceiveInterval);
}
handler.setAutoStartup(this.autoStartup);
return handler;
}

41
spring-messaging/src/main/java/org/springframework/messaging/simp/config/StompEndpointRegistration.java
View File

@@ -1,41 +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.messaging.simp.config;
import org.springframework.web.socket.server.HandshakeHandler;
import org.springframework.web.socket.server.config.SockJsServiceRegistration;
/**
* Provides methods for configuring a STOMP protocol handler including enabling SockJS
* fallback options.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
public interface StompEndpointRegistration {
/**
* Configure the HandshakeHandler to use.
*/
StompEndpointRegistration setHandshakeHandler(HandshakeHandler handshakeHandler);
/**
* Enable SockJS fallback options.
*/
SockJsServiceRegistration withSockJS();
}

33
spring-messaging/src/main/java/org/springframework/messaging/simp/config/StompEndpointRegistry.java
View File

@@ -1,33 +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.messaging.simp.config;
/**
* Provides methods for configuring STOMP protocol handlers at specific URL paths.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
public interface StompEndpointRegistry {
/**
* Expose a STOMP endpoint at the specified URL path (or paths_.
*/
StompEndpointRegistration addEndpoint(String... paths);
}

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

@@ -1,257 +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.messaging.simp.config;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.websocket.SubProtocolWebSocketHandler;
import org.springframework.messaging.simp.SimpMessageSendingOperations;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.handler.*;
import org.springframework.messaging.support.channel.AbstractSubscribableChannel;
import org.springframework.messaging.support.channel.ExecutorSubscribableChannel;
import org.springframework.messaging.support.converter.*;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.ClassUtils;
import org.springframework.util.MimeTypeUtils;
import org.springframework.web.servlet.HandlerMapping;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurationSupport;
import org.springframework.web.servlet.handler.AbstractHandlerMapping;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.config.SockJsServiceRegistration;
import java.util.ArrayList;
import java.util.List;
/**
* Configuration support for broker-backed messaging over WebSocket using a higher-level
* messaging sub-protocol such as STOMP. This class can either be extended directly
* or its configuration can also be customized in a callback style via
* {@link EnableWebSocketMessageBroker @EnableWebSocketMessageBroker} and
* {@link WebSocketMessageBrokerConfigurer}.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
public abstract class WebSocketMessageBrokerConfigurationSupport {
private static final boolean jackson2Present =
ClassUtils.isPresent("com.fasterxml.jackson.databind.ObjectMapper", WebMvcConfigurationSupport.class.getClassLoader()) &&
ClassUtils.isPresent("com.fasterxml.jackson.core.JsonGenerator", WebMvcConfigurationSupport.class.getClassLoader());
private MessageBrokerConfigurer messageBrokerConfigurer;
// WebSocket configuration including message channels to/from the application
@Bean
public HandlerMapping brokerWebSocketHandlerMapping() {
ServletStompEndpointRegistry registry = new ServletStompEndpointRegistry(
subProtocolWebSocketHandler(), userSessionRegistry(), brokerDefaultSockJsTaskScheduler());
registerStompEndpoints(registry);
AbstractHandlerMapping hm = registry.getHandlerMapping();
hm.setOrder(1);
return hm;
}
@Bean
public WebSocketHandler subProtocolWebSocketHandler() {
SubProtocolWebSocketHandler wsHandler = new SubProtocolWebSocketHandler(webSocketRequestChannel());
webSocketResponseChannel().subscribe(wsHandler);
return wsHandler;
}
@Bean
public UserSessionRegistry userSessionRegistry() {
return new DefaultUserSessionRegistry();
}
/**
* The default TaskScheduler to use if none is configured via
* {@link SockJsServiceRegistration#setTaskScheduler(org.springframework.scheduling.TaskScheduler)}, i.e.
* <pre class="code">
* &#064;Configuration
* &#064;EnableWebSocketMessageBroker
* public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
*
* public void registerStompEndpoints(StompEndpointRegistry registry) {
* registry.addEndpoint("/stomp").withSockJS().setTaskScheduler(myScheduler());
* }
*
* // ...
*
* }
* </pre>
*/
@Bean
public ThreadPoolTaskScheduler brokerDefaultSockJsTaskScheduler() {
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.setThreadNamePrefix("BrokerSockJS-");
return scheduler;
}
protected void registerStompEndpoints(StompEndpointRegistry registry) {
}
@Bean
public AbstractSubscribableChannel webSocketRequestChannel() {
return new ExecutorSubscribableChannel(webSocketRequestChannelExecutor());
}
@Bean
public ThreadPoolTaskExecutor webSocketRequestChannelExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setThreadNamePrefix("WebSocketRequestChannel-");
return executor;
}
@Bean
public AbstractSubscribableChannel webSocketResponseChannel() {
return new ExecutorSubscribableChannel(webSocketResponseChannelExecutor());
}
@Bean
public ThreadPoolTaskExecutor webSocketResponseChannelExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setThreadNamePrefix("WebSocketResponseChannel-");
return executor;
}
// Handling of messages by the application
@Bean
public SimpAnnotationMethodMessageHandler annotationMethodMessageHandler() {
SimpAnnotationMethodMessageHandler handler =
new SimpAnnotationMethodMessageHandler(brokerMessagingTemplate(), webSocketResponseChannel());
handler.setDestinationPrefixes(getMessageBrokerConfigurer().getApplicationDestinationPrefixes());
handler.setMessageConverter(simpMessageConverter());
webSocketRequestChannel().subscribe(handler);
return handler;
}
@Bean
public AbstractBrokerMessageHandler simpleBrokerMessageHandler() {
AbstractBrokerMessageHandler handler = getMessageBrokerConfigurer().getSimpleBroker();
if (handler == null) {
return noopBroker;
}
else {
webSocketRequestChannel().subscribe(handler);
brokerChannel().subscribe(handler);
return handler;
}
}
@Bean
public AbstractBrokerMessageHandler stompBrokerRelayMessageHandler() {
AbstractBrokerMessageHandler handler = getMessageBrokerConfigurer().getStompBrokerRelay();
if (handler == null) {
return noopBroker;
}
else {
webSocketRequestChannel().subscribe(handler);
brokerChannel().subscribe(handler);
return handler;
}
}
protected final MessageBrokerConfigurer getMessageBrokerConfigurer() {
if (this.messageBrokerConfigurer == null) {
MessageBrokerConfigurer configurer = new MessageBrokerConfigurer(webSocketResponseChannel());
configureMessageBroker(configurer);
this.messageBrokerConfigurer = configurer;
}
return this.messageBrokerConfigurer;
}
protected void configureMessageBroker(MessageBrokerConfigurer configurer) {
}
@Bean
public UserDestinationMessageHandler userDestinationMessageHandler() {
UserDestinationMessageHandler handler = new UserDestinationMessageHandler(
brokerMessagingTemplate(), userDestinationResolver());
webSocketRequestChannel().subscribe(handler);
brokerChannel().subscribe(handler);
return handler;
}
@Bean
public SimpMessageSendingOperations brokerMessagingTemplate() {
SimpMessagingTemplate template = new SimpMessagingTemplate(brokerChannel());
String userDestinationPrefix = getMessageBrokerConfigurer().getUserDestinationPrefix();
if (userDestinationPrefix != null) {
template.setUserDestinationPrefix(userDestinationPrefix);
}
template.setMessageConverter(simpMessageConverter());
return template;
}
@Bean
public AbstractSubscribableChannel brokerChannel() {
return new ExecutorSubscribableChannel(); // synchronous
}
@Bean
public CompositeMessageConverter simpMessageConverter() {
DefaultContentTypeResolver contentTypeResolver = new DefaultContentTypeResolver();
List<MessageConverter> converters = new ArrayList<MessageConverter>();
if (jackson2Present) {
converters.add(new MappingJackson2MessageConverter());
contentTypeResolver.setDefaultMimeType(MimeTypeUtils.APPLICATION_JSON);
}
converters.add(new StringMessageConverter());
converters.add(new ByteArrayMessageConverter());
return new CompositeMessageConverter(converters, contentTypeResolver);
}
@Bean
public UserDestinationResolver userDestinationResolver() {
DefaultUserDestinationResolver resolver = new DefaultUserDestinationResolver(userSessionRegistry());
String prefix = getMessageBrokerConfigurer().getUserDestinationPrefix();
if (prefix != null) {
resolver.setUserDestinationPrefix(prefix);
}
return resolver;
}
private static final AbstractBrokerMessageHandler noopBroker = new AbstractBrokerMessageHandler(null) {
@Override
protected void startInternal() {
}
@Override
protected void stopInternal() {
}
@Override
protected void handleMessageInternal(Message<?> message) {
}
};
}

40
spring-messaging/src/main/java/org/springframework/messaging/simp/config/WebSocketMessageBrokerConfigurer.java
View File

@@ -1,40 +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.messaging.simp.config;
/**
* Defines callback methods to configure broker-backed messaging over WebSocket via
* {@link EnableWebSocketMessageBroker @EnableWebSocketMessageBroker}.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
public interface WebSocketMessageBrokerConfigurer {
/**
* Configure STOMP protocol handling over WebSocket at a specific URL.
*/
void registerStompEndpoints(StompEndpointRegistry registry);
/**
* Configure message broker options.
*/
void configureMessageBroker(MessageBrokerConfigurer configurer);
}

14
spring-messaging/src/main/java/org/springframework/messaging/simp/handler/SimpAnnotationMethodMessageHandler.java
View File

@@ -79,7 +79,7 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
private final SimpMessageSendingOperations brokerTemplate;
private final SimpMessageSendingOperations webSocketResponseTemplate;
private final SimpMessageSendingOperations clientMessagingTemplate;
private MessageConverter messageConverter;
@@ -90,15 +90,15 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
/**
* @param brokerTemplate a messaging template to send application messages to the broker
* @param webSocketResponseChannel the channel for messages to WebSocket clients
* @param clientOutboundChannel the channel for messages to clients (e.g. WebSocket clients)
*/
public SimpAnnotationMethodMessageHandler(SimpMessageSendingOperations brokerTemplate,
MessageChannel webSocketResponseChannel) {
MessageChannel clientOutboundChannel) {
Assert.notNull(brokerTemplate, "brokerTemplate is required");
Assert.notNull(webSocketResponseChannel, "webSocketReplyChannel is required");
Assert.notNull(clientOutboundChannel, "clientOutboundChannel is required");
this.brokerTemplate = brokerTemplate;
this.webSocketResponseTemplate = new SimpMessagingTemplate(webSocketResponseChannel);
this.clientMessagingTemplate = new SimpMessagingTemplate(clientOutboundChannel);
Collection<MessageConverter> converters = new ArrayList<MessageConverter>();
converters.add(new StringMessageConverter());
@@ -117,7 +117,7 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
public void setMessageConverter(MessageConverter converter) {
this.messageConverter = converter;
if (converter != null) {
((AbstractMessageSendingTemplate<?>) this.webSocketResponseTemplate).setMessageConverter(converter);
((AbstractMessageSendingTemplate<?>) this.clientMessagingTemplate).setMessageConverter(converter);
}
}
@@ -194,7 +194,7 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
// Annotation-based return value types
handlers.add(new SendToMethodReturnValueHandler(this.brokerTemplate, true));
handlers.add(new SubscriptionMethodReturnValueHandler(this.webSocketResponseTemplate));
handlers.add(new SubscriptionMethodReturnValueHandler(this.clientMessagingTemplate));
// custom return value types
handlers.addAll(getCustomReturnValueHandlers());

251
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompProtocolHandler.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.messaging.simp.stomp;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.security.Principal;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.handler.websocket.SubProtocolHandler;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.simp.handler.UserSessionRegistry;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;
/**
* A {@link SubProtocolHandler} for STOMP that supports versions 1.0, 1.1, and 1.2 of the
* STOMP specification.
*
* @author Rossen Stoyanchev
* @author Andy Wilkinson
* @since 4.0
*/
public class StompProtocolHandler implements SubProtocolHandler {
/**
* The name of the header set on the CONNECTED frame indicating the name of the user
* authenticated on the WebSocket session.
*/
public static final String CONNECTED_USER_HEADER = "user-name";
private static final Log logger = LogFactory.getLog(StompProtocolHandler.class);
private final StompDecoder stompDecoder = new StompDecoder();
private final StompEncoder stompEncoder = new StompEncoder();
private UserSessionRegistry userSessionRegistry;
/**
* Provide a registry with which to register active user session ids.
*
* @see {@link org.springframework.messaging.simp.handler.UserDestinationMessageHandler}
*/
public void setUserSessionRegistry(UserSessionRegistry registry) {
this.userSessionRegistry = registry;
}
/**
* @return the configured UserSessionRegistry.
*/
public UserSessionRegistry getUserSessionRegistry() {
return this.userSessionRegistry;
}
@Override
public List<String> getSupportedProtocols() {
return Arrays.asList("v10.stomp", "v11.stomp", "v12.stomp");
}
/**
* Handle incoming WebSocket messages from clients.
*/
public void handleMessageFromClient(WebSocketSession session, WebSocketMessage webSocketMessage,
MessageChannel outputChannel) {
Message<?> message;
try {
Assert.isInstanceOf(TextMessage.class, webSocketMessage);
String payload = ((TextMessage)webSocketMessage).getPayload();
ByteBuffer byteBuffer = ByteBuffer.wrap(payload.getBytes(Charset.forName("UTF-8")));
message = this.stompDecoder.decode(byteBuffer);
}
catch (Throwable error) {
logger.error("Failed to parse STOMP frame, WebSocket message payload: ", error);
sendErrorMessage(session, error);
return;
}
try {
StompHeaderAccessor headers = StompHeaderAccessor.wrap(message);
if (SimpMessageType.HEARTBEAT.equals(headers.getMessageType())) {
logger.trace("Received heartbeat from client session=" + session.getId());
}
else {
logger.trace("Received message from client session=" + session.getId());
}
headers.setSessionId(session.getId());
headers.setUser(session.getPrincipal());
message = MessageBuilder.withPayload(message.getPayload()).setHeaders(headers).build();
outputChannel.send(message);
}
catch (Throwable t) {
logger.error("Terminating STOMP session due to failure to send message: ", t);
sendErrorMessage(session, t);
}
}
protected void sendErrorMessage(WebSocketSession session, Throwable error) {
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.ERROR);
headers.setMessage(error.getMessage());
Message<byte[]> message = MessageBuilder.withPayload(new byte[0]).setHeaders(headers).build();
String payload = new String(this.stompEncoder.encode(message), Charset.forName("UTF-8"));
try {
session.sendMessage(new TextMessage(payload));
}
catch (Throwable t) {
// ignore
}
}
/**
* Handle STOMP messages going back out to WebSocket clients.
*/
@SuppressWarnings("unchecked")
@Override
public void handleMessageToClient(WebSocketSession session, Message<?> message) {
StompHeaderAccessor headers = StompHeaderAccessor.wrap(message);
if (headers.getMessageType() == SimpMessageType.CONNECT_ACK) {
StompHeaderAccessor connectedHeaders = StompHeaderAccessor.create(StompCommand.CONNECTED);
connectedHeaders.setVersion(getVersion(headers));
connectedHeaders.setHeartbeat(0, 0); // no heart-beat support with simple broker
headers = connectedHeaders;
}
else if (SimpMessageType.MESSAGE.equals(headers.getMessageType())) {
headers.updateStompCommandAsServerMessage();
}
if (headers.getCommand() == StompCommand.CONNECTED) {
afterStompSessionConnected(headers, session);
}
if (StompCommand.MESSAGE.equals(headers.getCommand()) && (headers.getSubscriptionId() == null)) {
logger.error("Ignoring message, no subscriptionId header: " + message);
return;
}
if (!(message.getPayload() instanceof byte[])) {
logger.error("Ignoring message, expected byte[] content: " + message);
return;
}
try {
message = MessageBuilder.withPayload(message.getPayload()).setHeaders(headers).build();
byte[] bytes = this.stompEncoder.encode((Message<byte[]>)message);
session.sendMessage(new TextMessage(new String(bytes, Charset.forName("UTF-8"))));
}
catch (Throwable t) {
sendErrorMessage(session, t);
}
finally {
if (StompCommand.ERROR.equals(headers.getCommand())) {
try {
session.close(CloseStatus.PROTOCOL_ERROR);
}
catch (IOException e) {
// Ignore
}
}
}
}
private String getVersion(StompHeaderAccessor connectAckHeaders) {
String name = StompHeaderAccessor.CONNECT_MESSAGE_HEADER;
Message<?> connectMessage = (Message<?>) connectAckHeaders.getHeader(name);
StompHeaderAccessor connectHeaders = StompHeaderAccessor.wrap(connectMessage);
Assert.notNull(connectMessage, "CONNECT_ACK does not contain original CONNECT " + connectAckHeaders);
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 + "'");
}
}
private void afterStompSessionConnected(StompHeaderAccessor headers, WebSocketSession session) {
Principal principal = session.getPrincipal();
if (principal != null) {
headers.setNativeHeader(CONNECTED_USER_HEADER, principal.getName());
if (this.userSessionRegistry != null) {
String suffix = session.getId();
this.userSessionRegistry.registerSessionId(principal.getName(), session.getId());
}
}
}
@Override
public String resolveSessionId(Message<?> message) {
StompHeaderAccessor headers = StompHeaderAccessor.wrap(message);
return headers.getSessionId();
}
@Override
public void afterSessionStarted(WebSocketSession session, MessageChannel outputChannel) {
}
@Override
public void afterSessionEnded(WebSocketSession session, CloseStatus closeStatus, MessageChannel outputChannel) {
if ((this.userSessionRegistry != null) && (session.getPrincipal() != null)) {
this.userSessionRegistry.unregisterSessionId(session.getPrincipal().getName(), session.getId());
}
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.DISCONNECT);
headers.setSessionId(session.getId());
Message<?> message = MessageBuilder.withPayload(new byte[0]).setHeaders(headers).build();
outputChannel.send(message);
}
}