Direct delegation to RSocketMessageHandler

Simplify handling by eliminating the use of a message channel. Instead
MessageHandlerAcceptor now extends from RSocketMessageHandler and
delegates directly to it.

See gh-21987

Author: rstoyanchev

spring-messaging/src/main/java/org/springframework/messaging/ReactiveMessageChannel.java

@@ -30,14 +30,12 @@
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.EmbeddedValueResolverAware;
-import org.springframework.context.SmartLifecycle;
 import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.core.codec.Decoder;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.format.support.DefaultFormattingConversionService;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.ReactiveSubscribableChannel;
 import org.springframework.messaging.handler.CompositeMessageCondition;
 import org.springframework.messaging.handler.DestinationPatternsMessageCondition;
 import org.springframework.messaging.handler.annotation.MessageMapping;
@@ -74,9 +72,11 @@
  * @see AbstractEncoderMethodReturnValueHandler
  */
 public class MessageMappingMessageHandler extends AbstractMethodMessageHandler<CompositeMessageCondition>
-		implements SmartLifecycle, EmbeddedValueResolverAware {
+		implements EmbeddedValueResolverAware {
 
-	private final ReactiveSubscribableChannel inboundChannel;
+	@Nullable
+	private Predicate<Class<?>> handlerPredicate =
+			beanType -> AnnotatedElementUtils.hasAnnotation(beanType, Controller.class);
 
 	private final List<Decoder<?>> decoders = new ArrayList<>();
 
@@ -90,20 +90,63 @@ public class MessageMappingMessageHandler extends AbstractMethodMessageHandler<C
 	@Nullable
 	private StringValueResolver valueResolver;
 
-	private volatile boolean running = false;
-
-	private final Object lifecycleMonitor = new Object();
-
 
-	public MessageMappingMessageHandler(ReactiveSubscribableChannel inboundChannel) {
-		Assert.notNull(inboundChannel, "`inboundChannel` is required");
-		this.inboundChannel = inboundChannel;
+	public MessageMappingMessageHandler() {
 		this.pathMatcher = new AntPathMatcher();
 		((AntPathMatcher) this.pathMatcher).setPathSeparator(".");
-		setHandlerPredicate(beanType -> AnnotatedElementUtils.hasAnnotation(beanType, Controller.class));
 	}
 
 
+	/**
+	 * Manually configure handlers to check for {@code @MessageMapping} methods.
+	 * <p><strong>Note:</strong> the given handlers are not required to be
+	 * annotated with {@code @Controller}. Consider also using
+	 * {@link #setAutoDetectDisabled()} if the intent is to use these handlers
+	 * instead of, and not in addition to {@code @Controller} classes. Or
+	 * alternatively use {@link #setHandlerPredicate(Predicate)} to select a
+	 * different set of beans based on a different criteria.
+	 * @param handlers the handlers to register
+	 * @see #setAutoDetectDisabled()
+	 * @see #setHandlerPredicate(Predicate)
+	 */
+	public void setHandlers(List<Object> handlers) {
+		for (Object handler : handlers) {
+			detectHandlerMethods(handler);
+		}
+		// Disable auto-detection..
+		this.handlerPredicate = null;
+	}
+
+	/**
+	 * Configure the predicate to use for selecting which Spring beans to check
+	 * for {@code @MessageMapping} methods. When set to {@code null},
+	 * auto-detection is turned off which is what
+	 * {@link #setAutoDetectDisabled()} does internally.
+	 * <p>The predicate used by default selects {@code @Controller} classes.
+	 * @see #setHandlers(List)
+	 * @see #setAutoDetectDisabled()
+	 */
+	public void setHandlerPredicate(@Nullable Predicate<Class<?>> handlerPredicate) {
+		this.handlerPredicate = handlerPredicate;
+	}
+
+	/**
+	 * Return the {@link #setHandlerPredicate configured} handler predicate.
+	 */
+	@Nullable
+	public Predicate<Class<?>> getHandlerPredicate() {
+		return this.handlerPredicate;
+	}
+
+	/**
+	 * Disable auto-detection of {@code @MessageMapping} methods, e.g. in
+	 * {@code @Controller}s, by setting {@link #setHandlerPredicate(Predicate)
+	 * setHandlerPredicate(null)}.
+	 */
+	public void setAutoDetectDisabled() {
+		this.handlerPredicate = null;
+	}
+
 	/**
 	 * Configure the decoders to use for incoming payloads.
 	 */
@@ -203,34 +246,9 @@ protected List<? extends HandlerMethodReturnValueHandler> initReturnValueHandler
 		return Collections.emptyList();
 	}
 
-
-	@Override
-	public final void start() {
-		synchronized (this.lifecycleMonitor) {
-			this.inboundChannel.subscribe(this);
-			this.running = true;
-		}
-	}
-
-	@Override
-	public final void stop() {
-		synchronized (this.lifecycleMonitor) {
-			this.running = false;
-			this.inboundChannel.unsubscribe(this);
-		}
-	}
-
-	@Override
-	public final void stop(Runnable callback) {
-		synchronized (this.lifecycleMonitor) {
-			stop();
-			callback.run();
-		}
-	}
-
 	@Override
-	public final boolean isRunning() {
-		return this.running;
+	protected Predicate<Class<?>> initHandlerPredicate() {
+		return this.handlerPredicate;
 	}
 
 

spring-messaging/src/main/java/org/springframework/messaging/ReactiveSubscribableChannel.java

@@ -30,6 +30,7 @@
 import org.apache.commons.logging.LogFactory;
 import reactor.core.publisher.Mono;
 
+import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -47,6 +48,7 @@
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
+import org.springframework.util.ObjectUtils;
 
 /**
  * Abstract base class for reactive HandlerMethod-based message handling.
@@ -61,7 +63,7 @@
  * @param <T> the type of the Object that contains information mapping information
  */
 public abstract class AbstractMethodMessageHandler<T>
-		implements ReactiveMessageHandler, ApplicationContextAware, InitializingBean {
+		implements ReactiveMessageHandler, ApplicationContextAware, InitializingBean, BeanNameAware {
 
 	/**
 	 * Bean name prefix for target beans behind scoped proxies. Used to exclude those
@@ -79,9 +81,6 @@ public abstract class AbstractMethodMessageHandler<T>
 	protected final Log logger = LogFactory.getLog(getClass());
 
 
-	@Nullable
-	private Predicate<Class<?>> handlerPredicate;
-
 	private ArgumentResolverConfigurer argumentResolverConfigurer = new ArgumentResolverConfigurer();
 
 	private ReturnValueHandlerConfigurer returnValueHandlerConfigurer = new ReturnValueHandlerConfigurer();
@@ -91,29 +90,14 @@ public abstract class AbstractMethodMessageHandler<T>
 	@Nullable
 	private ApplicationContext applicationContext;
 
+	@Nullable
+	private String beanName;
+
 	private final Map<T, HandlerMethod> handlerMethods = new LinkedHashMap<>(64);
 
 	private final MultiValueMap<String, T> destinationLookup = new LinkedMultiValueMap<>(64);
 
 
-	/**
-	 * Configure a predicate to decide if which beans in the Spring context
-	 * should be checked to see if they have message handling methods.
-	 * <p>By default this is not set and sub-classes should configure it in
-	 * order to enable auto-detection of message handling methods.
-	 */
-	public void setHandlerPredicate(@Nullable Predicate<Class<?>> handlerPredicate) {
-		this.handlerPredicate = handlerPredicate;
-	}
-
-	/**
-	 * Return the {@link #setHandlerPredicate configured} handler predicate.
-	 */
-	@Nullable
-	public Predicate<Class<?>> getHandlerPredicate() {
-		return this.handlerPredicate;
-	}
-
 	/**
 	 * Configure custom resolvers for handler method arguments.
 	 */
@@ -170,6 +154,16 @@ public ApplicationContext getApplicationContext() {
 		return this.applicationContext;
 	}
 
+	@Override
+	public void setBeanName(String name) {
+		this.beanName = name;
+	}
+
+	public String getBeanName() {
+		return this.beanName != null ? this.beanName :
+				getClass().getSimpleName() + "@" + ObjectUtils.getIdentityHexString(this);
+	}
+
 	/**
 	 * Subclasses can invoke this method to populate the MessagingAdviceBean cache
 	 * (e.g. to support "global" {@code @MessageExceptionHandler}).
@@ -234,8 +228,9 @@ private void initHandlerMethods() {
 			logger.warn("No ApplicationContext available for detecting beans with message handling methods.");
 			return;
 		}
-		if (this.handlerPredicate == null) {
-			logger.warn("'handlerPredicate' not configured: no auto-detection of message handling methods.");
+		Predicate<Class<?>> handlerPredicate = initHandlerPredicate();
+		if (handlerPredicate == null) {
+			logger.warn("[" + getBeanName() + "] No auto-detection of handler methods (e.g. in @Controller).");
 			return;
 		}
 		for (String beanName : this.applicationContext.getBeanNamesForType(Object.class)) {
@@ -250,13 +245,21 @@ private void initHandlerMethods() {
 						logger.debug("Could not resolve target class for bean with name '" + beanName + "'", ex);
 					}
 				}
-				if (beanType != null && this.handlerPredicate.test(beanType)) {
+				if (beanType != null && handlerPredicate.test(beanType)) {
 					detectHandlerMethods(beanName);
 				}
 			}
 		}
 	}
 
+	/**
+	 * Return the predicate to use to check whether a given Spring bean should
+	 * be introspected for message handling methods. If {@code null} is
+	 * returned, auto-detection is effectively disabled.
+	 */
+	@Nullable
+	protected abstract Predicate<Class<?>> initHandlerPredicate();
+
 	/**
 	 * Detect if the given handler has any methods that can handle messages and if
 	 * so register it with the extracted mapping information.