diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/SessionSubscriptionRegistration.java b/spring-websocket/src/main/java/org/springframework/web/messaging/SessionSubscriptionRegistration.java deleted file mode 100644 index 0fe7ee3ade..0000000000 --- a/spring-websocket/src/main/java/org/springframework/web/messaging/SessionSubscriptionRegistration.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * Copyright 2002-2013 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.web.messaging; - -import java.util.Set; - -/** - * @author Rossen Stoyanchev - * @since 4.0 - */ -public interface SessionSubscriptionRegistration { - - - String getSessionId(); - - void addSubscription(String destination, String subscriptionId); - - /** - * @param subscriptionId the subscription to remove - * @return the destination to which the subscriptionId was registered, or {@code null} - * if no matching subscriptionId was found - */ - String removeSubscription(String subscriptionId); - - Set getSubscriptionsByDestination(String destination); - - Set getDestinations(); - -} diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/SessionSubscriptionRegistry.java b/spring-websocket/src/main/java/org/springframework/web/messaging/SessionSubscriptionRegistry.java deleted file mode 100644 index 852b745e8e..0000000000 --- a/spring-websocket/src/main/java/org/springframework/web/messaging/SessionSubscriptionRegistry.java +++ /dev/null @@ -1,38 +0,0 @@ -/* - * Copyright 2002-2013 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.web.messaging; - -import java.util.Set; - - -/** - * @author Rossen Stoyanchev - * @since 4.0 - */ -public interface SessionSubscriptionRegistry { - - SessionSubscriptionRegistration getRegistration(String sessionId); - - SessionSubscriptionRegistration getOrCreateRegistration(String sessionId); - - SessionSubscriptionRegistration removeRegistration(String sessionId); - - Set getSessionSubscriptions(String sessionId, String destination); - - Set getRegistrationsByDestination(String destination); - -} diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/service/broker/AbstractSubscriptionRegistry.java b/spring-websocket/src/main/java/org/springframework/web/messaging/service/broker/AbstractSubscriptionRegistry.java new file mode 100644 index 0000000000..8b25a99e56 --- /dev/null +++ b/spring-websocket/src/main/java/org/springframework/web/messaging/service/broker/AbstractSubscriptionRegistry.java @@ -0,0 +1,108 @@ +/* + * Copyright 2002-2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.web.messaging.service.broker; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.messaging.Message; +import org.springframework.util.MultiValueMap; +import org.springframework.web.messaging.MessageType; +import org.springframework.web.messaging.support.WebMessageHeaderAccesssor; + + +/** + * @author Rossen Stoyanchev + * @since 4.0 + */ +public abstract class AbstractSubscriptionRegistry implements SubscriptionRegistry { + + protected final Log logger = LogFactory.getLog(getClass()); + + + @Override + public void addSubscription(Message message) { + WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.wrap(message); + if (!MessageType.SUBSCRIBE.equals(headers.getMessageType())) { + logger.error("Expected SUBSCRIBE message: " + message); + return; + } + String sessionId = headers.getSessionId(); + if (sessionId == null) { + logger.error("Ignoring subscription. No sessionId in message: " + message); + return; + } + String subscriptionId = headers.getSubscriptionId(); + if (subscriptionId == null) { + logger.error("Ignoring subscription. No subscriptionId in message: " + message); + return; + } + String destination = headers.getDestination(); + if (destination == null) { + logger.error("Ignoring destination. No destination in message: " + message); + return; + } + addSubscriptionInternal(sessionId, subscriptionId, destination, message); + } + + protected abstract void addSubscriptionInternal(String sessionId, String subscriptionId, + String destination, Message message); + + @Override + public void removeSubscription(Message message) { + WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.wrap(message); + if (!MessageType.UNSUBSCRIBE.equals(headers.getMessageType())) { + logger.error("Expected UNSUBSCRIBE message: " + message); + return; + } + String sessionId = headers.getSessionId(); + if (sessionId == null) { + logger.error("Ignoring subscription. No sessionId in message: " + message); + return; + } + String subscriptionId = headers.getSubscriptionId(); + if (subscriptionId == null) { + logger.error("Ignoring subscription. No subscriptionId in message: " + message); + return; + } + removeSubscriptionInternal(sessionId, subscriptionId, message); + } + + protected abstract void removeSubscriptionInternal(String sessionId, String subscriptionId, Message message); + + @Override + public void removeSessionSubscriptions(String sessionId) { + } + + @Override + public MultiValueMap findSubscriptions(Message message) { + WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.wrap(message); + if (!MessageType.MESSAGE.equals(headers.getMessageType())) { + logger.error("Unexpected message type: " + message); + return null; + } + String destination = headers.getDestination(); + if (destination == null) { + logger.error("Ignoring destination. No destination in message: " + message); + return null; + } + return findSubscriptionsInternal(destination, message); + } + + protected abstract MultiValueMap findSubscriptionsInternal( + String destination, Message message); + +} diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/service/broker/DefaultSubscriptionRegistry.java b/spring-websocket/src/main/java/org/springframework/web/messaging/service/broker/DefaultSubscriptionRegistry.java new file mode 100644 index 0000000000..fffe05b84c --- /dev/null +++ b/spring-websocket/src/main/java/org/springframework/web/messaging/service/broker/DefaultSubscriptionRegistry.java @@ -0,0 +1,240 @@ +/* + * Copyright 2002-2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.web.messaging.service.broker; + +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArraySet; + +import org.springframework.messaging.Message; +import org.springframework.util.AntPathMatcher; +import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MultiValueMap; + + +/** + * @author Rossen Stoyanchev + * @since 4.0 + */ +public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry { + + private final DestinationCache destinationCache = new DestinationCache(); + + private final SessionSubscriptionRegistry subscriptionRegistry = new SessionSubscriptionRegistry(); + + private AntPathMatcher pathMatcher = new AntPathMatcher(); + + + /** + * @param pathMatcher the pathMatcher to set + */ + public void setPathMatcher(AntPathMatcher pathMatcher) { + this.pathMatcher = pathMatcher; + } + + public AntPathMatcher getPathMatcher() { + return this.pathMatcher; + } + + @Override + protected void addSubscriptionInternal(String sessionId, String subsId, String destination, Message message) { + SessionSubscriptionInfo info = this.subscriptionRegistry.addSubscription(sessionId, subsId, destination); + if (!this.pathMatcher.isPattern(destination)) { + this.destinationCache.mapToDestination(destination, info); + } + } + + @Override + protected void removeSubscriptionInternal(String sessionId, String subscriptionId, Message message) { + SessionSubscriptionInfo info = this.subscriptionRegistry.getSubscriptions(sessionId); + if (info != null) { + String destination = info.removeSubscription(subscriptionId); + if (info.getSubscriptions(destination) == null) { + this.destinationCache.unmapFromDestination(destination, info); + } + } + } + + @Override + public void removeSessionSubscriptions(String sessionId) { + SessionSubscriptionInfo info = this.subscriptionRegistry.removeSubscriptions(sessionId); + this.destinationCache.removeSessionSubscriptions(info); + } + + @Override + protected MultiValueMap findSubscriptionsInternal(String destination, Message message) { + MultiValueMap result = this.destinationCache.getSubscriptions(destination); + if (result.isEmpty()) { + result = new LinkedMultiValueMap(); + for (SessionSubscriptionInfo info : this.subscriptionRegistry.getAllSubscriptions()) { + for (String destinationPattern : info.getDestinations()) { + if (this.pathMatcher.match(destinationPattern, destination)) { + for (String subscriptionId : info.getSubscriptions(destinationPattern)) { + result.add(info.sessionId, subscriptionId); + } + } + } + } + } + return result; + } + + + /** + * Provide direct lookup of session subscriptions by destination (for non-pattern destinations). + */ + private static class DestinationCache { + + // destination -> .. + private final Map> subscriptionsByDestination = + new ConcurrentHashMap>(); + + private final Object monitor = new Object(); + + + public void mapToDestination(String destination, SessionSubscriptionInfo info) { + synchronized (monitor) { + Set registrations = this.subscriptionsByDestination.get(destination); + if (registrations == null) { + registrations = new CopyOnWriteArraySet(); + this.subscriptionsByDestination.put(destination, registrations); + } + registrations.add(info); + } + } + + public void unmapFromDestination(String destination, SessionSubscriptionInfo info) { + synchronized (monitor) { + Set infos = this.subscriptionsByDestination.get(destination); + if (infos != null) { + infos.remove(info); + if (infos.isEmpty()) { + this.subscriptionsByDestination.remove(destination); + } + } + } + } + + public void removeSessionSubscriptions(SessionSubscriptionInfo info) { + for (String destination : info.getDestinations()) { + unmapFromDestination(destination, info); + } + } + + public MultiValueMap getSubscriptions(String destination) { + MultiValueMap result = new LinkedMultiValueMap(); + Set infos = this.subscriptionsByDestination.get(destination); + if (infos != null) { + for (SessionSubscriptionInfo info : infos) { + Set subscriptions = info.getSubscriptions(destination); + if (subscriptions != null) { + for (String subscription : subscriptions) { + result.add(info.getSessionId(), subscription); + } + } + } + } + return result; + } + } + + /** + * Provide access to session subscriptions by sessionId. + */ + private static class SessionSubscriptionRegistry { + + private final Map sessions = + new ConcurrentHashMap(); + + + public SessionSubscriptionInfo getSubscriptions(String sessionId) { + return this.sessions.get(sessionId); + } + + public Collection getAllSubscriptions() { + return this.sessions.values(); + } + + public SessionSubscriptionInfo addSubscription(String sessionId, String subscriptionId, String destination) { + SessionSubscriptionInfo info = this.sessions.get(sessionId); + if (info == null) { + info = new SessionSubscriptionInfo(sessionId); + this.sessions.put(sessionId, info); + } + info.addSubscription(subscriptionId, destination); + return info; + } + + public SessionSubscriptionInfo removeSubscriptions(String sessionId) { + return this.sessions.remove(sessionId); + } + } + + /** + * Hold subscriptions for a session. + */ + private static class SessionSubscriptionInfo { + + private final String sessionId; + + private final Map> subscriptions = new HashMap>(4); + + + public SessionSubscriptionInfo(String sessionId) { + this.sessionId = sessionId; + } + + public String getSessionId() { + return this.sessionId; + } + + public Set getDestinations() { + return this.subscriptions.keySet(); + } + + public Set getSubscriptions(String destination) { + return this.subscriptions.get(destination); + } + + public void addSubscription(String subscriptionId, String destination) { + Set subs = this.subscriptions.get(destination); + if (subs == null) { + subs = new HashSet(4); + this.subscriptions.put(destination, subs); + } + subs.add(subscriptionId); + } + + public String removeSubscription(String subscriptionId) { + for (String destination : this.subscriptions.keySet()) { + Set subscriptionIds = this.subscriptions.get(destination); + if (subscriptionIds.remove(subscriptionId)) { + if (subscriptionIds.isEmpty()) { + this.subscriptions.remove(destination); + } + return destination; + } + } + return null; + } + } + +} diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/service/SimpleBrokerWebMessageHandler.java b/spring-websocket/src/main/java/org/springframework/web/messaging/service/broker/SimpleBrokerWebMessageHandler.java similarity index 56% rename from spring-websocket/src/main/java/org/springframework/web/messaging/service/SimpleBrokerWebMessageHandler.java rename to spring-websocket/src/main/java/org/springframework/web/messaging/service/broker/SimpleBrokerWebMessageHandler.java index 651ed4b870..9198abf3dd 100644 --- a/spring-websocket/src/main/java/org/springframework/web/messaging/service/SimpleBrokerWebMessageHandler.java +++ b/spring-websocket/src/main/java/org/springframework/web/messaging/service/broker/SimpleBrokerWebMessageHandler.java @@ -14,21 +14,18 @@ * limitations under the License. */ -package org.springframework.web.messaging.service; +package org.springframework.web.messaging.service.broker; import java.util.Arrays; import java.util.Collection; -import java.util.Set; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; +import org.springframework.util.MultiValueMap; import org.springframework.web.messaging.MessageType; -import org.springframework.web.messaging.SessionSubscriptionRegistration; -import org.springframework.web.messaging.SessionSubscriptionRegistry; -import org.springframework.web.messaging.support.CachingSessionSubscriptionRegistry; -import org.springframework.web.messaging.support.DefaultSessionSubscriptionRegistry; +import org.springframework.web.messaging.service.AbstractWebMessageHandler; import org.springframework.web.messaging.support.WebMessageHeaderAccesssor; @@ -38,23 +35,22 @@ import org.springframework.web.messaging.support.WebMessageHeaderAccesssor; */ public class SimpleBrokerWebMessageHandler extends AbstractWebMessageHandler { - private final MessageChannel clientChannel; + private final MessageChannel outboundChannel; - private SessionSubscriptionRegistry subscriptionRegistry= - new CachingSessionSubscriptionRegistry(new DefaultSessionSubscriptionRegistry()); + private SubscriptionRegistry subscriptionRegistry = new DefaultSubscriptionRegistry(); /** - * @param clientChannel the channel to which messages for clients should be sent + * @param outboundChannel the channel to which messages for clients should be sent * @param observable an Observable to use to manage subscriptions */ - public SimpleBrokerWebMessageHandler(MessageChannel clientChannel) { - Assert.notNull(clientChannel, "clientChannel is required"); - this.clientChannel = clientChannel; + public SimpleBrokerWebMessageHandler(MessageChannel outboundChannel) { + Assert.notNull(outboundChannel, "outboundChannel is required"); + this.outboundChannel = outboundChannel; } - public void setSubscriptionRegistry(SessionSubscriptionRegistry subscriptionRegistry) { + public void setSubscriptionRegistry(SubscriptionRegistry subscriptionRegistry) { Assert.notNull(subscriptionRegistry, "subscriptionRegistry is required"); this.subscriptionRegistry = subscriptionRegistry; } @@ -71,13 +67,16 @@ public class SimpleBrokerWebMessageHandler extends AbstractWebMessageHandler { logger.debug("Subscribe " + message); } - WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.wrap(message); - String sessionId = headers.getSessionId(); - String subscriptionId = headers.getSubscriptionId(); - String destination = headers.getDestination(); + this.subscriptionRegistry.addSubscription(message); - SessionSubscriptionRegistration registration = this.subscriptionRegistry.getOrCreateRegistration(sessionId); - registration.addSubscription(destination, subscriptionId); + // TODO: need a way to communicate back if subscription was successfully created or + // not in which case an ERROR should be sent back and close the connection + // http://stomp.github.io/stomp-specification-1.2.html#SUBSCRIBE + } + + @Override + protected void handleUnsubscribe(Message message) { + this.subscriptionRegistry.removeSubscription(message); } @Override @@ -89,29 +88,24 @@ public class SimpleBrokerWebMessageHandler extends AbstractWebMessageHandler { String destination = WebMessageHeaderAccesssor.wrap(message).getDestination(); - Set registrations = - this.subscriptionRegistry.getRegistrationsByDestination(destination); + MultiValueMap subscriptions = this.subscriptionRegistry.findSubscriptions(message); - if (registrations == null) { - return; - } - - for (SessionSubscriptionRegistration registration : registrations) { - for (String subscriptionId : registration.getSubscriptionsByDestination(destination)) { + for (String sessionId : subscriptions.keySet()) { + for (String subscriptionId : subscriptions.get(sessionId)) { WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.wrap(message); - headers.setSessionId(registration.getSessionId()); + headers.setSessionId(sessionId); headers.setSubscriptionId(subscriptionId); Message clientMessage = MessageBuilder.withPayload( message.getPayload()).copyHeaders(headers.toMap()).build(); try { - this.clientChannel.send(clientMessage); + this.outboundChannel.send(clientMessage); } catch (Throwable ex) { logger.error("Failed to send message to destination=" + destination + - ", sessionId=" + registration.getSessionId() + ", subscriptionId=" + subscriptionId, ex); + ", sessionId=" + sessionId + ", subscriptionId=" + subscriptionId, ex); } } } @@ -120,7 +114,7 @@ public class SimpleBrokerWebMessageHandler extends AbstractWebMessageHandler { @Override public void handleDisconnect(Message message) { String sessionId = WebMessageHeaderAccesssor.wrap(message).getSessionId(); - this.subscriptionRegistry.removeRegistration(sessionId); + this.subscriptionRegistry.removeSessionSubscriptions(sessionId); } } diff --git a/spring-websocket/src/test/java/org/springframework/web/messaging/support/DefaultSessionSubscriptionRegistryTests.java b/spring-websocket/src/main/java/org/springframework/web/messaging/service/broker/SubscriptionRegistry.java similarity index 59% rename from spring-websocket/src/test/java/org/springframework/web/messaging/support/DefaultSessionSubscriptionRegistryTests.java rename to spring-websocket/src/main/java/org/springframework/web/messaging/service/broker/SubscriptionRegistry.java index c7021f3c35..868966d29d 100644 --- a/spring-websocket/src/test/java/org/springframework/web/messaging/support/DefaultSessionSubscriptionRegistryTests.java +++ b/spring-websocket/src/main/java/org/springframework/web/messaging/service/broker/SubscriptionRegistry.java @@ -14,22 +14,24 @@ * limitations under the License. */ -package org.springframework.web.messaging.support; +package org.springframework.web.messaging.service.broker; -import org.springframework.web.messaging.SessionSubscriptionRegistry; +import org.springframework.messaging.Message; +import org.springframework.util.MultiValueMap; /** - * Test fixture for {@link DefaultSessionSubscriptionRegistry}. - * * @author Rossen Stoyanchev + * @since 4.0 */ -public class DefaultSessionSubscriptionRegistryTests extends AbstractSessionSubscriptionRegistryTests { +public interface SubscriptionRegistry { + void addSubscription(Message subscribeMessage); - @Override - protected SessionSubscriptionRegistry createSessionSubscriptionRegistry() { - return new DefaultSessionSubscriptionRegistry(); - } + void removeSubscription(Message unsubscribeMessage); + + void removeSessionSubscriptions(String sessionId); + + MultiValueMap findSubscriptions(Message message); } diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/AnnotationWebMessageHandler.java b/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/AnnotationWebMessageHandler.java index 6a4489050f..7c1669d16e 100644 --- a/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/AnnotationWebMessageHandler.java +++ b/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/AnnotationWebMessageHandler.java @@ -57,9 +57,9 @@ import org.springframework.web.method.HandlerMethodSelector; public class AnnotationWebMessageHandler extends AbstractWebMessageHandler implements ApplicationContextAware, InitializingBean { - private final MessageChannel clientChannel; + private final MessageChannel inboundChannel; - private final MessageChannel brokerChannel; + private final MessageChannel outboundChannel; private List messageConverters; @@ -79,11 +79,15 @@ public class AnnotationWebMessageHandler extends AbstractWebMessageHandler private ReturnValueHandlerComposite returnValueHandlers = new ReturnValueHandlerComposite(); - public AnnotationWebMessageHandler(MessageChannel clientChannel, MessageChannel brokerChannel) { - Assert.notNull(clientChannel, "clientChannel is required"); - Assert.notNull(brokerChannel, "brokerChannel is required"); - this.clientChannel = clientChannel; - this.brokerChannel = brokerChannel; + /** + * @param inboundChannel a channel for processing incoming messages from clients + * @param outboundChannel a channel for messages going out to clients + */ + public AnnotationWebMessageHandler(MessageChannel inboundChannel, MessageChannel outboundChannel) { + Assert.notNull(inboundChannel, "inboundChannel is required"); + Assert.notNull(outboundChannel, "outboundChannel is required"); + this.inboundChannel = inboundChannel; + this.outboundChannel = outboundChannel; } public void setMessageConverters(List converters) { @@ -105,11 +109,11 @@ public class AnnotationWebMessageHandler extends AbstractWebMessageHandler initHandlerMethods(); - this.argumentResolvers.addResolver(new MessageChannelArgumentResolver(this.brokerChannel)); + this.argumentResolvers.addResolver(new MessageChannelArgumentResolver(this.inboundChannel)); this.argumentResolvers.addResolver(new MessageBodyArgumentResolver(this.messageConverters)); - this.returnValueHandlers.addHandler(new MessageReturnValueHandler(this.clientChannel)); - this.returnValueHandlers.addHandler(new PayloadReturnValueHandler(this.clientChannel)); + this.returnValueHandlers.addHandler(new MessageReturnValueHandler(this.outboundChannel)); + this.returnValueHandlers.addHandler(new PayloadReturnValueHandler(this.outboundChannel)); } protected void initHandlerMethods() { diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/MessageChannelArgumentResolver.java b/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/MessageChannelArgumentResolver.java index a13398b93a..d1fd783b0e 100644 --- a/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/MessageChannelArgumentResolver.java +++ b/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/MessageChannelArgumentResolver.java @@ -28,12 +28,12 @@ import org.springframework.util.Assert; */ public class MessageChannelArgumentResolver implements ArgumentResolver { - private MessageChannel messageBrokerChannel; + private MessageChannel inboundChannel; - public MessageChannelArgumentResolver(MessageChannel messageBrokerChannel) { - Assert.notNull(messageBrokerChannel, "messageBrokerChannel is required"); - this.messageBrokerChannel = messageBrokerChannel; + public MessageChannelArgumentResolver(MessageChannel inboundChannel) { + Assert.notNull(inboundChannel, "inboundChannel is required"); + this.inboundChannel = inboundChannel; } @Override @@ -43,7 +43,7 @@ public class MessageChannelArgumentResolver implements ArgumentResolver { @Override public Object resolveArgument(MethodParameter parameter, Message message) throws Exception { - return this.messageBrokerChannel; + return this.inboundChannel; } } diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/MessageReturnValueHandler.java b/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/MessageReturnValueHandler.java index d148feea36..4e49820cc2 100644 --- a/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/MessageReturnValueHandler.java +++ b/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/MessageReturnValueHandler.java @@ -30,12 +30,12 @@ import org.springframework.web.messaging.support.WebMessageHeaderAccesssor; */ public class MessageReturnValueHandler implements ReturnValueHandler { - private MessageChannel clientChannel; + private MessageChannel outboundChannel; - public MessageReturnValueHandler(MessageChannel clientChannel) { - Assert.notNull(clientChannel, "clientChannel is required"); - this.clientChannel = clientChannel; + public MessageReturnValueHandler(MessageChannel outboundChannel) { + Assert.notNull(outboundChannel, "outboundChannel is required"); + this.outboundChannel = outboundChannel; } @Override @@ -49,7 +49,7 @@ public class MessageReturnValueHandler implements ReturnValueHandler { public void handleReturnValue(Object returnValue, MethodParameter returnType, Message message) throws Exception { - Assert.notNull(this.clientChannel, "No clientChannel to send messages to"); + Assert.notNull(this.outboundChannel, "No clientChannel to send messages to"); Message returnMessage = (Message) returnValue; if (message == null) { @@ -70,7 +70,7 @@ public class MessageReturnValueHandler implements ReturnValueHandler { returnMessage = MessageBuilder.withPayload( returnMessage.getPayload()).copyHeaders(returnHeaders.toMap()).build(); - this.clientChannel.send(returnMessage); + this.outboundChannel.send(returnMessage); } } diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/PayloadReturnValueHandler.java b/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/PayloadReturnValueHandler.java index 1c6c8d6708..e910d85fe2 100644 --- a/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/PayloadReturnValueHandler.java +++ b/spring-websocket/src/main/java/org/springframework/web/messaging/service/method/PayloadReturnValueHandler.java @@ -30,12 +30,12 @@ import org.springframework.web.messaging.support.WebMessageHeaderAccesssor; */ public class PayloadReturnValueHandler implements ReturnValueHandler { - private MessageChannel clientChannel; + private MessageChannel outboundChannel; - public PayloadReturnValueHandler(MessageChannel clientChannel) { - Assert.notNull(clientChannel, "clientChannel is required"); - this.clientChannel = clientChannel; + public PayloadReturnValueHandler(MessageChannel outboundChannel) { + Assert.notNull(outboundChannel, "outboundChannel is required"); + this.outboundChannel = outboundChannel; } @Override @@ -47,7 +47,7 @@ public class PayloadReturnValueHandler implements ReturnValueHandler { public void handleReturnValue(Object returnValue, MethodParameter returnType, Message message) throws Exception { - Assert.notNull(this.clientChannel, "No clientChannel to send messages to"); + Assert.notNull(this.outboundChannel, "No outboundChannel to send messages to"); if (returnValue == null) { return; @@ -63,7 +63,7 @@ public class PayloadReturnValueHandler implements ReturnValueHandler { Message returnMessage = MessageBuilder.withPayload( returnValue).copyHeaders(returnHeaders.toMap()).build(); - this.clientChannel.send(returnMessage); + this.outboundChannel.send(returnMessage); } } diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompHeaderAccessor.java b/spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompHeaderAccessor.java index 9899abca20..da7e8cb37f 100644 --- a/spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompHeaderAccessor.java +++ b/spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompHeaderAccessor.java @@ -97,7 +97,7 @@ public class StompHeaderAccessor extends WebMessageHeaderAccesssor { if (contentType != null) { super.setContentType(MediaType.parseMediaType(contentType)); } - if (StompCommand.SUBSCRIBE.equals(getStompCommand())) { + if (StompCommand.SUBSCRIBE.equals(getStompCommand()) || StompCommand.UNSUBSCRIBE.equals(getStompCommand())) { if (getFirstNativeHeader(STOMP_ID) != null) { super.setSubscriptionId(getFirstNativeHeader(STOMP_ID)); } diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompRelayWebMessageHandler.java b/spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompRelayWebMessageHandler.java index 3eda11708f..29db5804d7 100644 --- a/spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompRelayWebMessageHandler.java +++ b/spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompRelayWebMessageHandler.java @@ -59,7 +59,7 @@ public class StompRelayWebMessageHandler extends AbstractWebMessageHandler imple private static final String STOMP_RELAY_SYSTEM_SESSION_ID = "stompRelaySystemSessionId"; - private MessageChannel clientChannel; + private MessageChannel outboundChannel; private String relayHost = "127.0.0.1"; @@ -85,11 +85,11 @@ public class StompRelayWebMessageHandler extends AbstractWebMessageHandler imple /** - * @param clientChannel the channel to which messages for clients should be sent. + * @param outboundChannel a channel for messages going out to clients */ - public StompRelayWebMessageHandler(MessageChannel clientChannel) { - Assert.notNull(clientChannel, "clientChannel is required"); - this.clientChannel = clientChannel; + public StompRelayWebMessageHandler(MessageChannel outboundChannel) { + Assert.notNull(outboundChannel, "outboundChannel is required"); + this.outboundChannel = outboundChannel; this.payloadConverter = new CompositeMessageConverter(null); } @@ -387,7 +387,7 @@ public class StompRelayWebMessageHandler extends AbstractWebMessageHandler imple } protected void sendMessageToClient(Message message) { - clientChannel.send(message); + outboundChannel.send(message); } private void sendError(String sessionId, String errorText) { diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompWebSocketHandler.java b/spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompWebSocketHandler.java index 844bb9b599..7533e72a17 100644 --- a/spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompWebSocketHandler.java +++ b/spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompWebSocketHandler.java @@ -29,15 +29,11 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.support.MessageBuilder; -import org.springframework.util.CollectionUtils; import org.springframework.web.messaging.MessageType; -import org.springframework.web.messaging.SessionSubscriptionRegistration; -import org.springframework.web.messaging.SessionSubscriptionRegistry; import org.springframework.web.messaging.converter.CompositeMessageConverter; import org.springframework.web.messaging.converter.MessageConverter; import org.springframework.web.messaging.stomp.StompCommand; import org.springframework.web.messaging.stomp.StompConversionException; -import org.springframework.web.messaging.support.DefaultSessionSubscriptionRegistry; import org.springframework.web.messaging.support.WebMessageHeaderAccesssor; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.TextMessage; @@ -63,8 +59,6 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement private final Map sessions = new ConcurrentHashMap(); - private SessionSubscriptionRegistry subscriptionRegistry = new DefaultSessionSubscriptionRegistry(); - private MessageConverter payloadConverter = new CompositeMessageConverter(null); @@ -86,10 +80,6 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement return this.stompMessageConverter; } - public void setSubscriptionRegistry(SessionSubscriptionRegistry subscriptionRegistry) { - this.subscriptionRegistry = subscriptionRegistry; - } - @Override public void afterConnectionEstablished(WebSocketSession session) throws Exception { @@ -179,35 +169,12 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement } protected void handleSubscribe(Message message) { - - // TODO: need a way to communicate back if subscription was successfully created or - // not in which case an ERROR should be sent back and close the connection - // http://stomp.github.io/stomp-specification-1.2.html#SUBSCRIBE - - StompHeaderAccessor headers = StompHeaderAccessor.wrap(message); - String sessionId = headers.getSessionId(); - String destination = headers.getDestination(); - - SessionSubscriptionRegistration registration = this.subscriptionRegistry.getOrCreateRegistration(sessionId); - registration.addSubscription(destination, headers.getSubscriptionId()); } protected void handleUnsubscribe(Message message) { - - StompHeaderAccessor headers = StompHeaderAccessor.wrap(message); - String sessionId = headers.getSessionId(); - String subscriptionId = headers.getSubscriptionId(); - - SessionSubscriptionRegistration registration = this.subscriptionRegistry.getRegistration(sessionId); - if (registration == null) { - logger.warn("Subscripton=" + subscriptionId + " for session=" + sessionId + " not found"); - return; - } - registration.removeSubscription(subscriptionId); } protected void handleDisconnect(Message message) { - } protected void sendErrorMessage(WebSocketSession session, Throwable error) { @@ -230,7 +197,6 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { this.sessions.remove(session.getId()); - this.subscriptionRegistry.removeRegistration(session.getId()); WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.create(MessageType.DISCONNECT); headers.setSessionId(session.getId()); @@ -254,25 +220,22 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement String sessionId = headers.getSessionId(); if (sessionId == null) { + // TODO: failed message delivery mechanism logger.error("No \"sessionId\" header in message: " + message); + return; } WebSocketSession session = this.sessions.get(sessionId); if (session == null) { - logger.error("Session not found: " + message); + // TODO: failed message delivery mechanism + logger.error("WebSocketSession not found for sessionId=" + sessionId); + return; } if (headers.getSubscriptionId() == null) { - String destination = headers.getDestination(); - Set subs = this.subscriptionRegistry.getSessionSubscriptions(sessionId, destination); - if (!CollectionUtils.isEmpty(subs)) { - // TODO: send to all subscriptions ids - headers.setSubscriptionId(subs.iterator().next()); - } - else { - logger.error("No subscription id: " + message); - return; - } + // TODO: failed message delivery mechanism + logger.error("No subscription id: " + message); + return; } byte[] payload; diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/support/CachingSessionSubscriptionRegistry.java b/spring-websocket/src/main/java/org/springframework/web/messaging/support/CachingSessionSubscriptionRegistry.java deleted file mode 100644 index 5b71138e92..0000000000 --- a/spring-websocket/src/main/java/org/springframework/web/messaging/support/CachingSessionSubscriptionRegistry.java +++ /dev/null @@ -1,190 +0,0 @@ -/* - * Copyright 2002-2013 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.web.messaging.support; - -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.CopyOnWriteArraySet; - -import org.springframework.util.Assert; -import org.springframework.web.messaging.SessionSubscriptionRegistration; -import org.springframework.web.messaging.SessionSubscriptionRegistry; - - -/** - * A decorator for a {@link SessionSubscriptionRegistry} that intercepts subscriptions - * being added and removed and maintains a lookup cache of registrations by destination. - * - * @author Rossen Stoyanchev - * @since 4.0 - */ -public class CachingSessionSubscriptionRegistry implements SessionSubscriptionRegistry { - - private final SessionSubscriptionRegistry delegate; - - private final DestinationCache destinationCache = new DestinationCache(); - - - public CachingSessionSubscriptionRegistry(SessionSubscriptionRegistry delegate) { - Assert.notNull(delegate, "delegate SessionSubscriptionRegistry is required"); - this.delegate = delegate; - } - - - @Override - public SessionSubscriptionRegistration getRegistration(String sessionId) { - SessionSubscriptionRegistration reg = this.delegate.getRegistration(sessionId); - return (reg != null) ? new CachingSessionSubscriptionRegistration(reg) : null; - } - - @Override - public SessionSubscriptionRegistration getOrCreateRegistration(String sessionId) { - return new CachingSessionSubscriptionRegistration(this.delegate.getOrCreateRegistration(sessionId)); - } - - @Override - public SessionSubscriptionRegistration removeRegistration(String sessionId) { - SessionSubscriptionRegistration registration = this.delegate.removeRegistration(sessionId); - if (registration != null) { - this.destinationCache.removeRegistration(registration); - } - return registration; - } - - @Override - public Set getSessionSubscriptions(String sessionId, String destination) { - return this.delegate.getSessionSubscriptions(sessionId, destination); - } - - @Override - public Set getRegistrationsByDestination(String destination) { - return this.destinationCache.getRegistrations(destination); - } - - - private static class DestinationCache { - - private final Map> cache = - new ConcurrentHashMap>(); - - private final Object monitor = new Object(); - - - public void mapRegistration(String destination, SessionSubscriptionRegistration registration) { - synchronized (monitor) { - Set registrations = this.cache.get(destination); - if (registrations == null) { - registrations = new CopyOnWriteArraySet(); - this.cache.put(destination, registrations); - } - registrations.add(registration); - } - } - - public void unmapRegistration(String destination, SessionSubscriptionRegistration registration) { - synchronized (monitor) { - Set registrations = this.cache.get(destination); - if (registrations != null) { - registrations.remove(registration); - if (registrations.isEmpty()) { - this.cache.remove(destination); - } - } - } - } - - private void removeRegistration(SessionSubscriptionRegistration registration) { - for (String destination : registration.getDestinations()) { - unmapRegistration(destination, registration); - } - } - - public Set getRegistrations(String destination) { - return this.cache.get(destination); - } - - @Override - public String toString() { - return "DestinationCache [cache=" + this.cache + "]"; - } - } - - private class CachingSessionSubscriptionRegistration implements SessionSubscriptionRegistration { - - private final SessionSubscriptionRegistration delegate; - - - public CachingSessionSubscriptionRegistration(SessionSubscriptionRegistration delegate) { - Assert.notNull(delegate, "delegate SessionSubscriptionRegistration is required"); - this.delegate = delegate; - } - - @Override - public String getSessionId() { - return this.delegate.getSessionId(); - } - - @Override - public void addSubscription(String destination, String subscriptionId) { - destinationCache.mapRegistration(destination, this); - this.delegate.addSubscription(destination, subscriptionId); - } - - @Override - public String removeSubscription(String subscriptionId) { - String destination = this.delegate.removeSubscription(subscriptionId); - if (destination != null && this.delegate.getSubscriptionsByDestination(destination) == null) { - destinationCache.unmapRegistration(destination, this); - } - return destination; - } - - @Override - public Set getSubscriptionsByDestination(String destination) { - return this.delegate.getSubscriptionsByDestination(destination); - } - - @Override - public Set getDestinations() { - return this.delegate.getDestinations(); - } - - @Override - public boolean equals(Object other) { - if (this == other) { - return true; - } - if (!(other instanceof CachingSessionSubscriptionRegistration)) { - return false; - } - CachingSessionSubscriptionRegistration otherType = (CachingSessionSubscriptionRegistration) other; - return this.delegate.equals(otherType.delegate); - } - - @Override - public int hashCode() { - return this.delegate.hashCode(); - } - - @Override - public String toString() { - return "CachingSessionSubscriptionRegistration [delegate=" + delegate + "]"; - } - } - -} diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/support/DefaultSessionSubscriptionRegistration.java b/spring-websocket/src/main/java/org/springframework/web/messaging/support/DefaultSessionSubscriptionRegistration.java deleted file mode 100644 index b0602835b2..0000000000 --- a/spring-websocket/src/main/java/org/springframework/web/messaging/support/DefaultSessionSubscriptionRegistration.java +++ /dev/null @@ -1,115 +0,0 @@ -/* - * Copyright 2002-2013 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.web.messaging.support; - -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - -import org.springframework.util.Assert; -import org.springframework.web.messaging.SessionSubscriptionRegistration; - - -/** - * A default implementation of SessionSubscriptionRegistration. Uses a map to keep track - * of subscriptions by destination. This implementation assumes that only one thread will - * access and update subscriptions at a time. - * - * @author Rossen Stoyanchev - * @since 4.0 - */ -public class DefaultSessionSubscriptionRegistration implements SessionSubscriptionRegistration { - - private final String sessionId; - - // destination -> subscriptionIds - private final Map> subscriptions = new HashMap>(4); - - - public DefaultSessionSubscriptionRegistration(String sessionId) { - Assert.notNull(sessionId, "sessionId is required"); - this.sessionId = sessionId; - } - - - public String getSessionId() { - return this.sessionId; - } - - @Override - public Set getDestinations() { - return this.subscriptions.keySet(); - } - - @Override - public void addSubscription(String destination, String subscriptionId) { - Assert.hasText(destination, "destination must not be empty"); - Assert.hasText(subscriptionId, "subscriptionId must not be empty"); - Set subs = this.subscriptions.get(destination); - if (subs == null) { - subs = new HashSet(4); - this.subscriptions.put(destination, subs); - } - subs.add(subscriptionId); - } - - @Override - public String removeSubscription(String subscriptionId) { - Assert.hasText(subscriptionId, "subscriptionId must not be empty"); - for (String destination : this.subscriptions.keySet()) { - Set subscriptionIds = this.subscriptions.get(destination); - if (subscriptionIds.remove(subscriptionId)) { - if (subscriptionIds.isEmpty()) { - this.subscriptions.remove(destination); - } - return destination; - } - } - return null; - } - - @Override - public Set getSubscriptionsByDestination(String destination) { - Assert.hasText(destination, "destination must not be empty"); - return this.subscriptions.get(destination); - } - - @Override - public boolean equals(Object other) { - if (this == other) { - return true; - } - if (!(other instanceof DefaultSessionSubscriptionRegistration)) { - return false; - } - DefaultSessionSubscriptionRegistration otherType = (DefaultSessionSubscriptionRegistration) other; - return this.sessionId.equals(otherType.sessionId); - } - - @Override - public int hashCode() { - return 31 + this.sessionId.hashCode(); - } - - @Override - public String toString() { - return "DefaultSessionSubscriptionRegistration [sessionId=" + this.sessionId - + ", subscriptions=" + this.subscriptions + "]"; - } - -} diff --git a/spring-websocket/src/main/java/org/springframework/web/messaging/support/DefaultSessionSubscriptionRegistry.java b/spring-websocket/src/main/java/org/springframework/web/messaging/support/DefaultSessionSubscriptionRegistry.java deleted file mode 100644 index 3f5bf75a42..0000000000 --- a/spring-websocket/src/main/java/org/springframework/web/messaging/support/DefaultSessionSubscriptionRegistry.java +++ /dev/null @@ -1,83 +0,0 @@ -/* - * Copyright 2002-2013 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.web.messaging.support; - -import java.util.HashSet; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; - -import org.springframework.web.messaging.SessionSubscriptionRegistration; -import org.springframework.web.messaging.SessionSubscriptionRegistry; - - -/** - * A default implementation of SessionSubscriptionRegistry. - * - * @author Rossen Stoyanchev - * @since 4.0 - */ -public class DefaultSessionSubscriptionRegistry implements SessionSubscriptionRegistry { - - // sessionId -> SessionSubscriptionRegistration - private final Map registrations = - new ConcurrentHashMap(); - - - @Override - public SessionSubscriptionRegistration getRegistration(String sessionId) { - return this.registrations.get(sessionId); - } - - @Override - public SessionSubscriptionRegistration getOrCreateRegistration(String sessionId) { - SessionSubscriptionRegistration registration = this.registrations.get(sessionId); - if (registration == null) { - registration = new DefaultSessionSubscriptionRegistration(sessionId); - this.registrations.put(sessionId, registration); - } - return registration; - } - - @Override - public SessionSubscriptionRegistration removeRegistration(String sessionId) { - return this.registrations.remove(sessionId); - } - - @Override - public Set getSessionSubscriptions(String sessionId, String destination) { - SessionSubscriptionRegistration registration = this.registrations.get(sessionId); - return (registration != null) ? registration.getSubscriptionsByDestination(destination) : null; - } - - /** - * The default implementation performs a lookup by destination on each registration. - * For a more efficient algorithm consider decorating an instance of this class with - * {@link CachingSessionSubscriptionRegistry}. - */ - @Override - public Set getRegistrationsByDestination(String destination) { - Set result = new HashSet(); - for (SessionSubscriptionRegistration r : this.registrations.values()) { - if (r.getSubscriptionsByDestination(destination) != null) { - result.add(r); - } - } - return result.isEmpty() ? null : result; - } - -} diff --git a/spring-websocket/src/test/java/org/springframework/web/messaging/service/SimpleBrokerWebMessageHandlerTests.java b/spring-websocket/src/test/java/org/springframework/web/messaging/service/SimpleBrokerWebMessageHandlerTests.java index 0353290684..0d9242907b 100644 --- a/spring-websocket/src/test/java/org/springframework/web/messaging/service/SimpleBrokerWebMessageHandlerTests.java +++ b/spring-websocket/src/test/java/org/springframework/web/messaging/service/SimpleBrokerWebMessageHandlerTests.java @@ -28,6 +28,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.MessageBuilder; import org.springframework.web.messaging.MessageType; +import org.springframework.web.messaging.service.broker.SimpleBrokerWebMessageHandler; import org.springframework.web.messaging.support.WebMessageHeaderAccesssor; import static org.junit.Assert.*; @@ -89,29 +90,29 @@ public class SimpleBrokerWebMessageHandlerTests { @Test public void subcribeDisconnectPublish() { - this.messageHandler.handleSubscribe(createSubscriptionMessage("sess1", "sub1", "/foo")); - this.messageHandler.handleSubscribe(createSubscriptionMessage("sess1", "sub2", "/foo")); - this.messageHandler.handleSubscribe(createSubscriptionMessage("sess1", "sub3", "/bar")); + String sess1 = "sess1"; + String sess2 = "sess2"; - this.messageHandler.handleSubscribe(createSubscriptionMessage("sess2", "sub1", "/foo")); - this.messageHandler.handleSubscribe(createSubscriptionMessage("sess2", "sub2", "/foo")); - this.messageHandler.handleSubscribe(createSubscriptionMessage("sess2", "sub3", "/bar")); + this.messageHandler.handleSubscribe(createSubscriptionMessage(sess1, "sub1", "/foo")); + this.messageHandler.handleSubscribe(createSubscriptionMessage(sess1, "sub2", "/foo")); + this.messageHandler.handleSubscribe(createSubscriptionMessage(sess1, "sub3", "/bar")); + + this.messageHandler.handleSubscribe(createSubscriptionMessage(sess2, "sub1", "/foo")); + this.messageHandler.handleSubscribe(createSubscriptionMessage(sess2, "sub2", "/foo")); + this.messageHandler.handleSubscribe(createSubscriptionMessage(sess2, "sub3", "/bar")); WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.create(MessageType.DISCONNECT); - headers.setSessionId("sess1"); + headers.setSessionId(sess1); Message message = MessageBuilder.withPayload(new byte[0]).copyHeaders(headers.toMap()).build(); this.messageHandler.handleDisconnect(message); this.messageHandler.handlePublish(createMessage("/foo", "message1")); this.messageHandler.handlePublish(createMessage("/bar", "message2")); - verify(this.clientChannel, times(6)).send(this.messageCaptor.capture()); - assertCapturedMessage("sess1", "sub1", "/foo"); - assertCapturedMessage("sess1", "sub2", "/foo"); - assertCapturedMessage("sess2", "sub1", "/foo"); - assertCapturedMessage("sess2", "sub2", "/foo"); - assertCapturedMessage("sess1", "sub3", "/bar"); - assertCapturedMessage("sess2", "sub3", "/bar"); + verify(this.clientChannel, times(3)).send(this.messageCaptor.capture()); + assertCapturedMessage(sess2, "sub1", "/foo"); + assertCapturedMessage(sess2, "sub2", "/foo"); + assertCapturedMessage(sess2, "sub3", "/bar"); } diff --git a/spring-websocket/src/test/java/org/springframework/web/messaging/service/broker/DefaultSubscriptionRegistryTests.java b/spring-websocket/src/test/java/org/springframework/web/messaging/service/broker/DefaultSubscriptionRegistryTests.java new file mode 100644 index 0000000000..55e4bfcc59 --- /dev/null +++ b/spring-websocket/src/test/java/org/springframework/web/messaging/service/broker/DefaultSubscriptionRegistryTests.java @@ -0,0 +1,242 @@ +/* + * Copyright 2002-2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.web.messaging.service.broker; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.MultiValueMap; +import org.springframework.web.messaging.MessageType; +import org.springframework.web.messaging.support.WebMessageHeaderAccesssor; + +import static org.junit.Assert.*; + + +/** + * Test fixture for {@link DefaultSubscriptionRegistry}. + * + * @author Rossen Stoyanchev + */ +public class DefaultSubscriptionRegistryTests { + + + private DefaultSubscriptionRegistry registry; + + + @Before + public void setup() { + this.registry = new DefaultSubscriptionRegistry(); + } + + + @Test + public void addSubscriptionInvalidInput() { + + String sessId = "sess01"; + String subsId = "subs01"; + String dest = "/foo"; + + this.registry.addSubscription(subscribeMessage(null, subsId, dest)); + assertEquals(0, this.registry.findSubscriptions(message(dest)).size()); + + this.registry.addSubscription(subscribeMessage(sessId, null, dest)); + assertEquals(0, this.registry.findSubscriptions(message(dest)).size()); + + this.registry.addSubscription(subscribeMessage(sessId, subsId, null)); + assertEquals(0, this.registry.findSubscriptions(message(dest)).size()); + } + + @Test + public void addSubscription() { + + String sessId = "sess01"; + String subsId = "subs01"; + String dest = "/foo"; + + this.registry.addSubscription(subscribeMessage(sessId, subsId, dest)); + MultiValueMap actual = this.registry.findSubscriptions(message(dest)); + + assertEquals("Expected one element " + actual, 1, actual.size()); + assertEquals(Arrays.asList(subsId), actual.get(sessId)); + } + + @Test + public void addSubscriptionOneSession() { + + String sessId = "sess01"; + List subscriptionIds = Arrays.asList("subs01", "subs02", "subs03"); + String dest = "/foo"; + + for (String subId : subscriptionIds) { + this.registry.addSubscription(subscribeMessage(sessId, subId, dest)); + } + + MultiValueMap actual = this.registry.findSubscriptions(message(dest)); + + assertEquals("Expected one element " + actual, 1, actual.size()); + assertEquals(subscriptionIds, sort(actual.get(sessId))); + } + + @Test + public void addSubscriptionMultipleSessions() { + + List sessIds = Arrays.asList("sess01", "sess02", "sess03"); + List subscriptionIds = Arrays.asList("subs01", "subs02", "subs03"); + String dest = "/foo"; + + for (String sessId : sessIds) { + for (String subsId : subscriptionIds) { + this.registry.addSubscription(subscribeMessage(sessId, subsId, dest)); + } + } + + MultiValueMap actual = this.registry.findSubscriptions(message(dest)); + + assertEquals("Expected three elements " + actual, 3, actual.size()); + assertEquals(subscriptionIds, sort(actual.get(sessIds.get(0)))); + assertEquals(subscriptionIds, sort(actual.get(sessIds.get(1)))); + assertEquals(subscriptionIds, sort(actual.get(sessIds.get(2)))); + } + + @Test + public void addSubscriptionWithDestinationPattern() { + + String sessId = "sess01"; + String subsId = "subs01"; + String destPattern = "/topic/PRICE.STOCK.*.IBM"; + String dest = "/topic/PRICE.STOCK.NASDAQ.IBM"; + + this.registry.addSubscription(subscribeMessage(sessId, subsId, destPattern)); + MultiValueMap actual = this.registry.findSubscriptions(message(dest)); + + assertEquals("Expected one element " + actual, 1, actual.size()); + assertEquals(Arrays.asList(subsId), actual.get(sessId)); + } + + @Test + public void addSubscriptionWithDestinationPatternRegex() { + + String sessId = "sess01"; + String subsId = "subs01"; + String destPattern = "/topic/PRICE.STOCK.*.{ticker:(IBM|MSFT)}"; + + this.registry.addSubscription(subscribeMessage(sessId, subsId, destPattern)); + Message message = message("/topic/PRICE.STOCK.NASDAQ.IBM"); + MultiValueMap actual = this.registry.findSubscriptions(message); + + assertEquals("Expected one element " + actual, 1, actual.size()); + assertEquals(Arrays.asList(subsId), actual.get(sessId)); + + message = message("/topic/PRICE.STOCK.NASDAQ.MSFT"); + actual = this.registry.findSubscriptions(message); + + assertEquals("Expected one element " + actual, 1, actual.size()); + assertEquals(Arrays.asList(subsId), actual.get(sessId)); + + message = message("/topic/PRICE.STOCK.NASDAQ.VMW"); + actual = this.registry.findSubscriptions(message); + + assertEquals("Expected no elements " + actual, 0, actual.size()); + } + + @Test + public void removeSubscription() { + + List sessIds = Arrays.asList("sess01", "sess02", "sess03"); + List subscriptionIds = Arrays.asList("subs01", "subs02", "subs03"); + String dest = "/foo"; + + for (String sessId : sessIds) { + for (String subsId : subscriptionIds) { + this.registry.addSubscription(subscribeMessage(sessId, subsId, dest)); + } + } + + this.registry.removeSubscription(unsubscribeMessage(sessIds.get(0), subscriptionIds.get(0))); + this.registry.removeSubscription(unsubscribeMessage(sessIds.get(0), subscriptionIds.get(1))); + this.registry.removeSubscription(unsubscribeMessage(sessIds.get(0), subscriptionIds.get(2))); + + MultiValueMap actual = this.registry.findSubscriptions(message(dest)); + + assertEquals("Expected three elements " + actual, 2, actual.size()); + assertEquals(subscriptionIds, sort(actual.get(sessIds.get(1)))); + assertEquals(subscriptionIds, sort(actual.get(sessIds.get(2)))); + } + + @Test + public void removeSessionSubscriptions() { + + List sessIds = Arrays.asList("sess01", "sess02", "sess03"); + List subscriptionIds = Arrays.asList("subs01", "subs02", "subs03"); + String dest = "/foo"; + + for (String sessId : sessIds) { + for (String subsId : subscriptionIds) { + this.registry.addSubscription(subscribeMessage(sessId, subsId, dest)); + } + } + + this.registry.removeSessionSubscriptions(sessIds.get(0)); + this.registry.removeSessionSubscriptions(sessIds.get(1)); + + MultiValueMap actual = this.registry.findSubscriptions(message(dest)); + + assertEquals("Expected three elements " + actual, 1, actual.size()); + assertEquals(subscriptionIds, sort(actual.get(sessIds.get(2)))); + } + + @Test + public void findSubscriptionsNoMatches() { + MultiValueMap actual = this.registry.findSubscriptions(message("/foo")); + assertEquals("Expected no elements " + actual, 0, actual.size()); + } + + + private Message subscribeMessage(String sessionId, String subscriptionId, String destination) { + WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.create(MessageType.SUBSCRIBE); + headers.setSessionId(sessionId); + headers.setSubscriptionId(subscriptionId); + if (destination != null) { + headers.setDestination(destination); + } + return MessageBuilder.withPayload("").copyHeaders(headers.toMap()).build(); + } + + private Message unsubscribeMessage(String sessionId, String subscriptionId) { + WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.create(MessageType.UNSUBSCRIBE); + headers.setSessionId(sessionId); + headers.setSubscriptionId(subscriptionId); + return MessageBuilder.withPayload("").copyHeaders(headers.toMap()).build(); + } + + private Message message(String destination) { + WebMessageHeaderAccesssor headers = WebMessageHeaderAccesssor.create(); + headers.setDestination(destination); + return MessageBuilder.withPayload("").copyHeaders(headers.toMap()).build(); + } + + private List sort(List list) { + Collections.sort(list); + return list; + } + +} diff --git a/spring-websocket/src/test/java/org/springframework/web/messaging/support/AbstractSessionSubscriptionRegistryTests.java b/spring-websocket/src/test/java/org/springframework/web/messaging/support/AbstractSessionSubscriptionRegistryTests.java deleted file mode 100644 index 55c0274803..0000000000 --- a/spring-websocket/src/test/java/org/springframework/web/messaging/support/AbstractSessionSubscriptionRegistryTests.java +++ /dev/null @@ -1,116 +0,0 @@ -/* - * Copyright 2002-2013 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.web.messaging.support; - -import java.util.Set; - -import org.junit.Before; -import org.junit.Test; -import org.springframework.web.messaging.SessionSubscriptionRegistration; -import org.springframework.web.messaging.SessionSubscriptionRegistry; - -import static org.junit.Assert.*; - - -/** - * A test fixture for {@link AbstractSessionSubscriptionRegistry}. - * - * @author Rossen Stoyanchev - */ -public abstract class AbstractSessionSubscriptionRegistryTests { - - protected SessionSubscriptionRegistry registry; - - - @Before - public void setup() { - this.registry = createSessionSubscriptionRegistry(); - } - - protected abstract SessionSubscriptionRegistry createSessionSubscriptionRegistry(); - - - @Test - public void getRegistration() { - String sessionId = "sess1"; - assertNull(this.registry.getRegistration(sessionId)); - - this.registry.getOrCreateRegistration(sessionId); - assertNotNull(this.registry.getRegistration(sessionId)); - assertEquals(sessionId, this.registry.getRegistration(sessionId).getSessionId()); - } - - @Test - public void getOrCreateRegistration() { - String sessionId = "sess1"; - assertNull(this.registry.getRegistration(sessionId)); - - SessionSubscriptionRegistration registration = this.registry.getOrCreateRegistration(sessionId); - assertEquals(registration, this.registry.getOrCreateRegistration(sessionId)); - } - - @Test - public void removeRegistration() { - String sessionId = "sess1"; - this.registry.getOrCreateRegistration(sessionId); - assertNotNull(this.registry.getRegistration(sessionId)); - assertEquals(sessionId, this.registry.getRegistration(sessionId).getSessionId()); - - this.registry.removeRegistration(sessionId); - assertNull(this.registry.getRegistration(sessionId)); - } - - @Test - public void getSessionSubscriptions() { - String sessionId = "sess1"; - SessionSubscriptionRegistration registration = this.registry.getOrCreateRegistration(sessionId); - registration.addSubscription("/foo", "sub1"); - registration.addSubscription("/foo", "sub2"); - - Set subscriptions = this.registry.getSessionSubscriptions(sessionId, "/foo"); - assertEquals("Wrong number of subscriptions " + subscriptions, 2, subscriptions.size()); - assertTrue(subscriptions.contains("sub1")); - assertTrue(subscriptions.contains("sub2")); - } - - @Test - public void getRegistrationsByDestination() { - - SessionSubscriptionRegistration reg1 = this.registry.getOrCreateRegistration("sess1"); - reg1.addSubscription("/foo", "sub1"); - - SessionSubscriptionRegistration reg2 = this.registry.getOrCreateRegistration("sess2"); - reg2.addSubscription("/foo", "sub1"); - - Set actual = this.registry.getRegistrationsByDestination("/foo"); - assertEquals(2, actual.size()); - assertTrue(actual.contains(reg1)); - assertTrue(actual.contains(reg2)); - - reg1.removeSubscription("sub1"); - - actual = this.registry.getRegistrationsByDestination("/foo"); - assertEquals("Invalid set of registrations " + actual, 1, actual.size()); - assertTrue(actual.contains(reg2)); - - reg2.removeSubscription("sub1"); - - actual = this.registry.getRegistrationsByDestination("/foo"); - assertNull("Unexpected registrations " + actual, actual); - } - -} diff --git a/spring-websocket/src/test/java/org/springframework/web/messaging/support/CachingSessionSubscriptionRegistryTests.java b/spring-websocket/src/test/java/org/springframework/web/messaging/support/CachingSessionSubscriptionRegistryTests.java deleted file mode 100644 index b2ecf87c91..0000000000 --- a/spring-websocket/src/test/java/org/springframework/web/messaging/support/CachingSessionSubscriptionRegistryTests.java +++ /dev/null @@ -1,35 +0,0 @@ -/* - * Copyright 2002-2013 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.web.messaging.support; - -import org.springframework.web.messaging.SessionSubscriptionRegistry; - - -/** - * Test fixture for {@link CachingSessionSubscriptionRegistry}. - * - * @author Rossen Stoyanchev - */ -public class CachingSessionSubscriptionRegistryTests extends AbstractSessionSubscriptionRegistryTests { - - - @Override - protected SessionSubscriptionRegistry createSessionSubscriptionRegistry() { - return new CachingSessionSubscriptionRegistry(new DefaultSessionSubscriptionRegistry()); - } - -} diff --git a/spring-websocket/src/test/java/org/springframework/web/messaging/support/DefaultSessionSubscriptionRegistrationTests.java b/spring-websocket/src/test/java/org/springframework/web/messaging/support/DefaultSessionSubscriptionRegistrationTests.java deleted file mode 100644 index 2fc7c481a2..0000000000 --- a/spring-websocket/src/test/java/org/springframework/web/messaging/support/DefaultSessionSubscriptionRegistrationTests.java +++ /dev/null @@ -1,82 +0,0 @@ -/* - * Copyright 2002-2013 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.web.messaging.support; - -import java.util.Set; - -import org.junit.Before; -import org.junit.Test; - -import static org.junit.Assert.*; - - -/** - * Test fixture for {@link DefaultSessionSubscriptionRegistration}. - * - * @author Rossen Stoyanchev - */ -public class DefaultSessionSubscriptionRegistrationTests { - - private DefaultSessionSubscriptionRegistration registration; - - - @Before - public void setup() { - this.registration = new DefaultSessionSubscriptionRegistration("sess1"); - } - - @Test - public void addSubscriptions() { - this.registration.addSubscription("/foo", "sub1"); - this.registration.addSubscription("/foo", "sub2"); - this.registration.addSubscription("/bar", "sub3"); - this.registration.addSubscription("/bar", "sub4"); - - assertSet(this.registration.getSubscriptionsByDestination("/foo"), 2, "sub1", "sub2"); - assertSet(this.registration.getSubscriptionsByDestination("/bar"), 2, "sub3", "sub4"); - assertSet(this.registration.getDestinations(), 2, "/foo", "/bar"); - } - - @Test - public void removeSubscriptions() { - this.registration.addSubscription("/foo", "sub1"); - this.registration.addSubscription("/foo", "sub2"); - this.registration.addSubscription("/bar", "sub3"); - this.registration.addSubscription("/bar", "sub4"); - - assertEquals("/foo", this.registration.removeSubscription("sub1")); - assertEquals("/foo", this.registration.removeSubscription("sub2")); - - assertNull(this.registration.getSubscriptionsByDestination("/foo")); - assertSet(this.registration.getDestinations(), 1, "/bar"); - - assertEquals("/bar", this.registration.removeSubscription("sub3")); - assertEquals("/bar", this.registration.removeSubscription("sub4")); - - assertNull(this.registration.getSubscriptionsByDestination("/bar")); - assertSet(this.registration.getDestinations(), 0); - } - - - private void assertSet(Set set, int size, String... elements) { - assertEquals("Wrong number of elements in " + set, size, set.size()); - for (String element : elements) { - assertTrue("Set does not contain element " + element, set.contains(element)); - } - } - -}