Refactor and polish spring-messaging

Remove base class for STOMP-related message handler classes
(AbstractSimpMessageHandler), polish subclasses and fix issues with
more significant updates to STOMP broker relay.

Introduce base class for SubscribableChannel implementations providing
consistent logging for all channel implementations.

Author: rstoyanchev

spring-messaging/src/main/java/org/springframework/messaging/simp/handler/AbstractSimpMessageHandler.java

@@ -34,7 +34,7 @@ public abstract class AbstractSubscriptionRegistry implements SubscriptionRegist
 
 
 	@Override
-	public void addSubscription(Message<?> message) {
+	public final void registerSubscription(Message<?> message) {
 		SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
 		if (!SimpMessageType.SUBSCRIBE.equals(headers.getMessageType())) {
 			logger.error("Expected SUBSCRIBE message: " + message);
@@ -55,14 +55,17 @@ public void addSubscription(Message<?> message) {
 			logger.error("Ignoring destination. No destination in message: " + message);
 			return;
 		}
+		if (logger.isDebugEnabled()) {
+			logger.debug("Subscribe request: " + message);
+		}
 		addSubscriptionInternal(sessionId, subscriptionId, destination, message);
 	}
 
 	protected abstract void addSubscriptionInternal(String sessionId, String subscriptionId,
 			String destination, Message<?> message);
 
 	@Override
-	public void removeSubscription(Message<?> message) {
+	public final void unregisterSubscription(Message<?> message) {
 		SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
 		if (!SimpMessageType.UNSUBSCRIBE.equals(headers.getMessageType())) {
 			logger.error("Expected UNSUBSCRIBE message: " + message);
@@ -78,17 +81,19 @@ public void removeSubscription(Message<?> message) {
 			logger.error("Ignoring subscription. No subscriptionId in message: " + message);
 			return;
 		}
+		if (logger.isDebugEnabled()) {
+			logger.debug("Unubscribe request: " + message);
+		}
 		removeSubscriptionInternal(sessionId, subscriptionId, message);
 	}
 
 	protected abstract void removeSubscriptionInternal(String sessionId, String subscriptionId, Message<?> message);
 
 	@Override
-	public void removeSessionSubscriptions(String sessionId) {
-	}
+	public abstract void unregisterAllSubscriptions(String sessionId);
 
 	@Override
-	public MultiValueMap<String, String> findSubscriptions(Message<?> message) {
+	public final MultiValueMap<String, String> findSubscriptions(Message<?> message) {
 		SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
 		if (!SimpMessageType.MESSAGE.equals(headers.getMessageType())) {
 			logger.error("Unexpected message type: " + message);
@@ -99,6 +104,9 @@ public MultiValueMap<String, String> findSubscriptions(Message<?> message) {
 			logger.error("Ignoring destination. No destination in message: " + message);
 			return null;
 		}
+		if (logger.isTraceEnabled()) {
+			logger.trace("Find subscriptions, destination=" + headers.getDestination());
+		}
 		return findSubscriptionsInternal(destination, message);
 	}
 

spring-messaging/src/main/java/org/springframework/messaging/simp/handler/AbstractSubscriptionRegistry.java

@@ -19,13 +19,14 @@
 import java.lang.annotation.Annotation;
 import java.lang.reflect.Method;
 import java.util.Arrays;
-import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.ApplicationContext;
@@ -34,6 +35,8 @@
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.handler.annotation.MessageMapping;
 import org.springframework.messaging.handler.annotation.support.MessageBodyArgumentResolver;
 import org.springframework.messaging.handler.annotation.support.MessageExceptionHandlerMethodResolver;
@@ -60,8 +63,9 @@
  * @author Rossen Stoyanchev
  * @since 4.0
  */
-public class AnnotationSimpMessageHandler extends AbstractSimpMessageHandler
-		implements ApplicationContextAware, InitializingBean {
+public class AnnotationSimpMessageHandler implements MessageHandler, ApplicationContextAware, InitializingBean {
+
+	private static final Log logger = LogFactory.getLog(AnnotationSimpMessageHandler.class);
 
 	private final MessageChannel outboundChannel;
 
@@ -104,11 +108,6 @@ public void setApplicationContext(ApplicationContext applicationContext) throws
 		this.applicationContext = applicationContext;
 	}
 
-	@Override
-	protected Collection<SimpMessageType> getSupportedMessageTypes() {
-		return Arrays.asList(SimpMessageType.MESSAGE, SimpMessageType.SUBSCRIBE, SimpMessageType.UNSUBSCRIBE);
-	}
-
 	@Override
 	public void afterPropertiesSet() {
 
@@ -183,18 +182,20 @@ protected HandlerMethod createHandlerMethod(Object handler, Method method) {
 	}
 
 	@Override
-	public void handlePublish(Message<?> message) {
-		handleMessageInternal(message, this.messageMethods);
-	}
+	public void handleMessage(Message<?> message) throws MessagingException {
 
-	@Override
-	public void handleSubscribe(Message<?> message) {
-		handleMessageInternal(message, this.subscribeMethods);
-	}
+		SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
+		SimpMessageType messageType = headers.getMessageType();
 
-	@Override
-	public void handleUnsubscribe(Message<?> message) {
-		handleMessageInternal(message, this.unsubscribeMethods);
+		if (SimpMessageType.MESSAGE.equals(messageType)) {
+			handleMessageInternal(message, this.messageMethods);
+		}
+		else if (SimpMessageType.SUBSCRIBE.equals(messageType)) {
+			handleMessageInternal(message, this.subscribeMethods);
+		}
+		else if (SimpMessageType.UNSUBSCRIBE.equals(messageType)) {
+			handleMessageInternal(message, this.unsubscribeMethods);
+		}
 	}
 
 	private void handleMessageInternal(final Message<?> message, Map<MappingInfo, HandlerMethod> handlerMethods) {

spring-messaging/src/main/java/org/springframework/messaging/simp/handler/AnnotationSimpMessageHandler.java

@@ -74,9 +74,14 @@ protected void removeSubscriptionInternal(String sessionId, String subscriptionI
 	}
 
 	@Override
-	public void removeSessionSubscriptions(String sessionId) {
+	public void unregisterAllSubscriptions(String sessionId) {
 		SessionSubscriptionInfo info = this.subscriptionRegistry.removeSubscriptions(sessionId);
-		this.destinationCache.removeSessionSubscriptions(info);
+		if (info != null) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Unregistering subscriptions for sessionId=" + sessionId);
+			}
+			this.destinationCache.removeSessionSubscriptions(info);
+		}
 	}
 
 	@Override