Payload encoding/decoding and handling refinements

See gh-21987

Author: rstoyanchev

spring-messaging/src/main/java/org/springframework/messaging/handler/HandlerMethod.java

@@ -298,7 +298,7 @@ public HandlerMethod createWithResolvedBean() {
 	 */
 	public String getShortLogMessage() {
 		int args = this.method.getParameterCount();
-		return getBeanType().getName() + "#" + this.method.getName() + "[" + args + " args]";
+		return getBeanType().getSimpleName() + "#" + this.method.getName() + "[" + args + " args]";
 	}
 
 

spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/support/reactive/MessageMappingMessageHandler.java

@@ -19,21 +19,25 @@
 import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Comparator;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.function.Predicate;
 
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 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;
@@ -51,58 +55,57 @@
 import org.springframework.validation.Validator;
 
 /**
- * Extension of {@link AbstractMethodMessageHandler} for
- * {@link MessageMapping @MessageMapping} methods.
+ * Extension of {@link AbstractMethodMessageHandler} for reactive, non-blocking
+ * handling of messages via {@link MessageMapping @MessageMapping} methods.
+ * By default such methods are detected in {@code @Controller} Spring beans but
+ * that can be changed via {@link #setHandlerPredicate(Predicate)}.
  *
- * <p>The payload of incoming messages is decoded through
- * {@link PayloadMethodArgumentResolver} using one of the configured
- * {@link #setDecoders(List)} decoders.
+ * <p>Payloads for incoming messages are decoded through the configured
+ * {@link #setDecoders(List)} decoders, with the help of
+ * {@link PayloadMethodArgumentResolver}.
  *
- * <p>The {@link #setEncoderReturnValueHandler encoderReturnValueHandler}
- * property must be set to encode and handle return values from
- * {@code @MessageMapping} methods.
+ * <p>There is no default handling for return values but
+ * {@link #setReturnValueHandlerConfigurer} can be used to configure custom
+ * return value handlers. Sub-classes may also override
+ * {@link #initReturnValueHandlers()} to set up default return value handlers.
  *
  * @author Rossen Stoyanchev
  * @since 5.2
+ * @see AbstractEncoderMethodReturnValueHandler
  */
 public class MessageMappingMessageHandler extends AbstractMethodMessageHandler<CompositeMessageCondition>
-		implements EmbeddedValueResolverAware {
+		implements SmartLifecycle, EmbeddedValueResolverAware {
 
-	private PathMatcher pathMatcher = new AntPathMatcher();
+	private final ReactiveSubscribableChannel inboundChannel;
 
 	private final List<Decoder<?>> decoders = new ArrayList<>();
 
 	@Nullable
 	private Validator validator;
 
-	@Nullable
-	private HandlerMethodReturnValueHandler encoderReturnValueHandler;
+	private PathMatcher pathMatcher;
 
 	private ConversionService conversionService = new DefaultFormattingConversionService();
 
 	@Nullable
 	private StringValueResolver valueResolver;
 
+	private volatile boolean running = false;
 
-	/**
-	 * Set the PathMatcher implementation to use for matching destinations
-	 * against configured destination patterns.
-	 * <p>By default, {@link AntPathMatcher} is used.
-	 */
-	public void setPathMatcher(PathMatcher pathMatcher) {
-		Assert.notNull(pathMatcher, "PathMatcher must not be null");
-		this.pathMatcher = pathMatcher;
-	}
+	private final Object lifecycleMonitor = new Object();
 
-	/**
-	 * Return the PathMatcher implementation to use for matching destinations.
-	 */
-	public PathMatcher getPathMatcher() {
-		return this.pathMatcher;
+
+	public MessageMappingMessageHandler(ReactiveSubscribableChannel inboundChannel) {
+		Assert.notNull(inboundChannel, "`inboundChannel` is required");
+		this.inboundChannel = inboundChannel;
+		this.pathMatcher = new AntPathMatcher();
+		((AntPathMatcher) this.pathMatcher).setPathSeparator(".");
+		setHandlerPredicate(beanType -> AnnotatedElementUtils.hasAnnotation(beanType, Controller.class));
 	}
 
+
 	/**
-	 * Configure the decoders to user for incoming payloads.
+	 * Configure the decoders to use for incoming payloads.
 	 */
 	public void setDecoders(List<? extends Decoder<?>> decoders) {
 		this.decoders.addAll(decoders);
@@ -115,14 +118,6 @@ public List<? extends Decoder<?>> getDecoders() {
 		return this.decoders;
 	}
 
-	/**
-	 * Return the configured Validator instance.
-	 */
-	@Nullable
-	public Validator getValidator() {
-		return this.validator;
-	}
-
 	/**
 	 * Set the Validator instance used for validating {@code @Payload} arguments.
 	 * @see org.springframework.validation.annotation.Validated
@@ -133,27 +128,28 @@ public void setValidator(@Nullable Validator validator) {
 	}
 
 	/**
-	 * Configure the return value handler that will encode response content.
-	 * Consider extending {@link AbstractEncoderMethodReturnValueHandler} which
-	 * provides the infrastructure to encode and all that's left is to somehow
-	 * handle the encoded content, e.g. by wrapping as a message and passing it
-	 * to something or sending it somewhere.
-	 * <p>By default this is not configured in which case payload/content return
-	 * values from {@code @MessageMapping} methods will remain unhandled.
-	 * @param encoderReturnValueHandler the return value handler to use
-	 * @see AbstractEncoderMethodReturnValueHandler
+	 * Return the configured Validator instance.
 	 */
-	public void setEncoderReturnValueHandler(@Nullable HandlerMethodReturnValueHandler encoderReturnValueHandler) {
-		this.encoderReturnValueHandler = encoderReturnValueHandler;
+	@Nullable
+	public Validator getValidator() {
+		return this.validator;
 	}
 
 	/**
-	 * Return the configured
-	 * {@link #setEncoderReturnValueHandler encoderReturnValueHandler}.
+	 * Set the PathMatcher implementation to use for matching destinations
+	 * against configured destination patterns.
+	 * <p>By default, {@link AntPathMatcher} is used with separator set to ".".
 	 */
-	@Nullable
-	public HandlerMethodReturnValueHandler getEncoderReturnValueHandler() {
-		return this.encoderReturnValueHandler;
+	public void setPathMatcher(PathMatcher pathMatcher) {
+		Assert.notNull(pathMatcher, "PathMatcher must not be null");
+		this.pathMatcher = pathMatcher;
+	}
+
+	/**
+	 * Return the PathMatcher implementation to use for matching destinations.
+	 */
+	public PathMatcher getPathMatcher() {
+		return this.pathMatcher;
 	}
 
 	/**
@@ -204,20 +200,40 @@ protected List<? extends HandlerMethodArgumentResolver> initArgumentResolvers()
 
 	@Override
 	protected List<? extends HandlerMethodReturnValueHandler> initReturnValueHandlers() {
-		List<HandlerMethodReturnValueHandler> handlers = new ArrayList<>();
-		handlers.addAll(getReturnValueHandlerConfigurer().getCustomHandlers());
-		if (this.encoderReturnValueHandler != null) {
-			handlers.add(this.encoderReturnValueHandler);
+		return Collections.emptyList();
+	}
+
+
+	@Override
+	public final void start() {
+		synchronized (this.lifecycleMonitor) {
+			this.inboundChannel.subscribe(this);
+			this.running = true;
 		}
-		return handlers;
 	}
 
+	@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
-	protected boolean isHandler(Class<?> beanType) {
-		return AnnotatedElementUtils.hasAnnotation(beanType, Controller.class);
+	public final boolean isRunning() {
+		return this.running;
 	}
 
+
 	@Override
 	protected CompositeMessageCondition getMappingForMethod(Method method, Class<?> handlerType) {
 		CompositeMessageCondition methodCondition = getCondition(method);

spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/support/reactive/PayloadMethodArgumentResolver.java

@@ -19,6 +19,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.function.Consumer;
 
 import org.apache.commons.logging.Log;
@@ -148,38 +149,47 @@ public boolean supportsParameter(MethodParameter parameter) {
 	 * @param message the message from which the content was extracted
 	 * @return a Mono with the result of argument resolution
 	 *
-	 * @see #extractPayloadContent(MethodParameter, Message)
+	 * @see #extractContent(MethodParameter, Message)
 	 * @see #getMimeType(Message)
 	 */
 	@Override
 	public final Mono<Object> resolveArgument(MethodParameter parameter, Message<?> message) {
+
 		Payload ann = parameter.getParameterAnnotation(Payload.class);
 		if (ann != null && StringUtils.hasText(ann.expression())) {
 			throw new IllegalStateException("@Payload SpEL expressions not supported by this resolver");
 		}
-		Publisher<DataBuffer> content = extractPayloadContent(parameter, message);
-		return decodeContent(parameter, message, ann == null || ann.required(), content, getMimeType(message));
+
+		MimeType mimeType = getMimeType(message);
+		mimeType = mimeType != null ? mimeType : MimeTypeUtils.APPLICATION_OCTET_STREAM;
+
+		Flux<DataBuffer> content = extractContent(parameter, message);
+		return decodeContent(parameter, message, ann == null || ann.required(), content, mimeType);
 	}
 
-	/**
-	 * Extract the content to decode from the message. By default, the message
-	 * payload is expected to be {@code Publisher<DataBuffer>}. Sub-classes can
-	 * override this method to change that assumption.
-	 * @param parameter the target method parameter we're decoding to
-	 * @param message the input message with the content
-	 * @return the content to decode
-	 */
 	@SuppressWarnings("unchecked")
-	protected Publisher<DataBuffer> extractPayloadContent(MethodParameter parameter, Message<?> message) {
-		Publisher<DataBuffer> content;
-		try {
-			content = (Publisher<DataBuffer>) message.getPayload();
+	private Flux<DataBuffer> extractContent(MethodParameter parameter, Message<?> message) {
+		Object payload = message.getPayload();
+		if (payload instanceof DataBuffer) {
+			return Flux.just((DataBuffer) payload);
 		}
-		catch (ClassCastException ex) {
-			throw new MethodArgumentResolutionException(
-					message, parameter, "Expected Publisher<DataBuffer> payload", ex);
+		if (payload instanceof Publisher) {
+			return Flux.from((Publisher<?>) payload).map(value -> {
+				if (value instanceof DataBuffer) {
+					return (DataBuffer) value;
+				}
+				String className = value.getClass().getName();
+				throw getUnexpectedPayloadError(message, parameter, "Publisher<" + className + ">");
+			});
 		}
-		return content;
+		return Flux.error(getUnexpectedPayloadError(message, parameter, payload.getClass().getName()));
+	}
+
+	private MethodArgumentResolutionException getUnexpectedPayloadError(
+			Message<?> message, MethodParameter parameter, String actualType) {
+
+		return new MethodArgumentResolutionException(message, parameter,
+				"Expected DataBuffer or Publisher<DataBuffer> for the Message payload, actual: " + actualType);
 	}
 
 	/**
@@ -206,7 +216,7 @@ else if (headerValue instanceof MimeType) {
 	}
 
 	private Mono<Object> decodeContent(MethodParameter parameter, Message<?> message,
-			boolean isContentRequired, Publisher<DataBuffer> content, @Nullable MimeType mimeType) {
+			boolean isContentRequired, Flux<DataBuffer> content, MimeType mimeType) {
 
 		ResolvableType targetType = ResolvableType.forMethodParameter(parameter);
 		Class<?> resolvedType = targetType.resolve();
@@ -215,19 +225,14 @@ private Mono<Object> decodeContent(MethodParameter parameter, Message<?> message
 		isContentRequired = isContentRequired || (adapter != null && !adapter.supportsEmpty());
 		Consumer<Object> validator = getValidator(message, parameter);
 
-		if (logger.isDebugEnabled()) {
-			logger.debug("Mime type:" + mimeType);
-		}
-		mimeType = mimeType != null ? mimeType : MimeTypeUtils.APPLICATION_OCTET_STREAM;
+		Map<String, Object> hints = Collections.emptyMap();
 
 		for (Decoder<?> decoder : this.decoders) {
 			if (decoder.canDecode(elementType, mimeType)) {
 				if (adapter != null && adapter.isMultiValue()) {
-					if (logger.isDebugEnabled()) {
-						logger.debug("0..N [" + elementType + "]");
-					}
-					Flux<?> flux = decoder.decode(content, elementType, mimeType, Collections.emptyMap());
-					flux = flux.onErrorResume(ex -> Flux.error(handleReadError(parameter, message, ex)));
+					Flux<?> flux = content
+							.concatMap(buffer -> decoder.decode(Mono.just(buffer), elementType, mimeType, hints))
+							.onErrorResume(ex -> Flux.error(handleReadError(parameter, message, ex)));
 					if (isContentRequired) {
 						flux = flux.switchIfEmpty(Flux.error(() -> handleMissingBody(parameter, message)));
 					}
@@ -237,12 +242,10 @@ private Mono<Object> decodeContent(MethodParameter parameter, Message<?> message
 					return Mono.just(adapter.fromPublisher(flux));
 				}
 				else {
-					if (logger.isDebugEnabled()) {
-						logger.debug("0..1 [" + elementType + "]");
-					}
 					// Single-value (with or without reactive type wrapper)
-					Mono<?> mono = decoder.decodeToMono(content, targetType, mimeType, Collections.emptyMap());
-					mono = mono.onErrorResume(ex -> Mono.error(handleReadError(parameter, message, ex)));
+					Mono<?> mono = decoder
+							.decodeToMono(content.next(), targetType, mimeType, hints)
+							.onErrorResume(ex -> Mono.error(handleReadError(parameter, message, ex)));
 					if (isContentRequired) {
 						mono = mono.switchIfEmpty(Mono.error(() -> handleMissingBody(parameter, message)));
 					}