Resolvers for destination vars and headers

See gh-21987

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

@@ -0,0 +1,235 @@
+/*
+ * Copyright 2002-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.handler.annotation.support.reactive;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.config.BeanExpressionContext;
+import org.springframework.beans.factory.config.BeanExpressionResolver;
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.core.MethodParameter;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.core.convert.TypeDescriptor;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.handler.annotation.ValueConstants;
+import org.springframework.messaging.handler.invocation.reactive.SyncHandlerMethodArgumentResolver;
+import org.springframework.util.ClassUtils;
+
+/**
+ * Abstract base class to resolve method arguments from a named value, e.g.
+ * message headers or destination variables. Named values could have one or more
+ * of a name, a required flag, and a default value.
+ *
+ * <p>Subclasses only need to define specific steps such as how to obtain named
+ * value details from a method parameter, how to resolve to argument values, or
+ * how to handle missing values.
+ *
+ *  <p>A default value string can contain ${...} placeholders and Spring
+ * Expression Language {@code #{...}} expressions which will be resolved if a
+ * {@link ConfigurableBeanFactory} is supplied to the class constructor.
+ *
+ * <p>A {@link ConversionService} is used to to convert resolved String argument
+ * value to the expected target method parameter type.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.2
+ */
+public abstract class AbstractNamedValueMethodArgumentResolver implements SyncHandlerMethodArgumentResolver {
+
+	private final ConversionService conversionService;
+
+	@Nullable
+	private final ConfigurableBeanFactory configurableBeanFactory;
+
+	@Nullable
+	private final BeanExpressionContext expressionContext;
+
+	private final Map<MethodParameter, NamedValueInfo> namedValueInfoCache = new ConcurrentHashMap<>(256);
+
+
+	/**
+	 * Constructor with a {@link ConversionService} and a {@link BeanFactory}.
+	 * @param conversionService conversion service for converting String values
+	 * to the target method parameter type
+	 * @param beanFactory a bean factory for resolving {@code ${...}}
+	 * placeholders and {@code #{...}} SpEL expressions in default values
+	 */
+	protected AbstractNamedValueMethodArgumentResolver(ConversionService conversionService,
+			@Nullable ConfigurableBeanFactory beanFactory) {
+
+		this.conversionService = conversionService;
+		this.configurableBeanFactory = beanFactory;
+		this.expressionContext = (beanFactory != null ? new BeanExpressionContext(beanFactory, null) : null);
+	}
+
+
+	@Override
+	public Object resolveArgumentValue(MethodParameter parameter, Message<?> message) {
+
+		NamedValueInfo namedValueInfo = getNamedValueInfo(parameter);
+		MethodParameter nestedParameter = parameter.nestedIfOptional();
+
+		Object resolvedName = resolveEmbeddedValuesAndExpressions(namedValueInfo.name);
+		if (resolvedName == null) {
+			throw new IllegalArgumentException(
+					"Specified name must not resolve to null: [" + namedValueInfo.name + "]");
+		}
+
+		Object arg = resolveArgumentInternal(nestedParameter, message, resolvedName.toString());
+		if (arg == null) {
+			if (namedValueInfo.defaultValue != null) {
+				arg = resolveEmbeddedValuesAndExpressions(namedValueInfo.defaultValue);
+			}
+			else if (namedValueInfo.required && !nestedParameter.isOptional()) {
+				handleMissingValue(namedValueInfo.name, nestedParameter, message);
+			}
+			arg = handleNullValue(namedValueInfo.name, arg, nestedParameter.getNestedParameterType());
+		}
+		else if ("".equals(arg) && namedValueInfo.defaultValue != null) {
+			arg = resolveEmbeddedValuesAndExpressions(namedValueInfo.defaultValue);
+		}
+
+		if (parameter != nestedParameter || !ClassUtils.isAssignableValue(parameter.getParameterType(), arg)) {
+			arg = this.conversionService.convert(arg, TypeDescriptor.forObject(arg), new TypeDescriptor(parameter));
+		}
+
+		return arg;
+	}
+
+	/**
+	 * Obtain the named value for the given method parameter.
+	 */
+	private NamedValueInfo getNamedValueInfo(MethodParameter parameter) {
+		NamedValueInfo namedValueInfo = this.namedValueInfoCache.get(parameter);
+		if (namedValueInfo == null) {
+			namedValueInfo = createNamedValueInfo(parameter);
+			namedValueInfo = updateNamedValueInfo(parameter, namedValueInfo);
+			this.namedValueInfoCache.put(parameter, namedValueInfo);
+		}
+		return namedValueInfo;
+	}
+
+	/**
+	 * Create the {@link NamedValueInfo} object for the given method parameter.
+	 * Implementations typically retrieve the method annotation by means of
+	 * {@link MethodParameter#getParameterAnnotation(Class)}.
+	 * @param parameter the method parameter
+	 * @return the named value information
+	 */
+	protected abstract NamedValueInfo createNamedValueInfo(MethodParameter parameter);
+
+	/**
+	 * Fall back on the parameter name from the class file if necessary and
+	 * replace {@link ValueConstants#DEFAULT_NONE} with null.
+	 */
+	private NamedValueInfo updateNamedValueInfo(MethodParameter parameter, NamedValueInfo info) {
+		String name = info.name;
+		if (info.name.isEmpty()) {
+			name = parameter.getParameterName();
+			if (name == null) {
+				Class<?> type = parameter.getParameterType();
+				throw new IllegalArgumentException(
+						"Name for argument of type [" + type.getName() + "] not specified, " +
+								"and parameter name information not found in class file either.");
+			}
+		}
+		return new NamedValueInfo(name, info.required,
+				ValueConstants.DEFAULT_NONE.equals(info.defaultValue) ? null : info.defaultValue);
+	}
+
+	/**
+	 * Resolve the given annotation-specified value,
+	 * potentially containing placeholders and expressions.
+	 */
+	@Nullable
+	private Object resolveEmbeddedValuesAndExpressions(String value) {
+		if (this.configurableBeanFactory == null || this.expressionContext == null) {
+			return value;
+		}
+		String placeholdersResolved = this.configurableBeanFactory.resolveEmbeddedValue(value);
+		BeanExpressionResolver exprResolver = this.configurableBeanFactory.getBeanExpressionResolver();
+		if (exprResolver == null) {
+			return value;
+		}
+		return exprResolver.evaluate(placeholdersResolved, this.expressionContext);
+	}
+
+	/**
+	 * Resolves the given parameter type and value name into an argument value.
+	 * @param parameter the method parameter to resolve to an argument value
+	 * @param message the current request
+	 * @param name the name of the value being resolved
+	 * @return the resolved argument. May be {@code null}
+	 */
+	@Nullable
+	protected abstract Object resolveArgumentInternal(MethodParameter parameter, Message<?> message, String name);
+
+	/**
+	 * Invoked when a value is required, but {@link #resolveArgumentInternal}
+	 * returned {@code null} and there is no default value. Sub-classes can
+	 * throw an appropriate exception for this case.
+	 * @param name the name for the value
+	 * @param parameter the target method parameter
+	 * @param message the message being processed
+	 */
+	protected abstract void handleMissingValue(String name, MethodParameter parameter, Message<?> message);
+
+	/**
+	 * One last chance to handle a possible null value.
+	 * Specifically for booleans method parameters, use {@link Boolean#FALSE}.
+	 * Also raise an ISE for primitive types.
+	 */
+	@Nullable
+	private Object handleNullValue(String name, @Nullable Object value, Class<?> paramType) {
+		if (value == null) {
+			if (Boolean.TYPE.equals(paramType)) {
+				return Boolean.FALSE;
+			}
+			else if (paramType.isPrimitive()) {
+				throw new IllegalStateException("Optional " + paramType + " parameter '" + name +
+						"' is present but cannot be translated into a null value due to being " +
+						"declared as a primitive type. Consider declaring it as object wrapper " +
+						"for the corresponding primitive type.");
+			}
+		}
+		return value;
+	}
+
+
+	/**
+	 * Represents a named value declaration.
+	 */
+	protected static class NamedValueInfo {
+
+		private final String name;
+
+		private final boolean required;
+
+		@Nullable
+		private final String defaultValue;
+
+		protected NamedValueInfo(String name, boolean required, @Nullable String defaultValue) {
+			this.name = name;
+			this.required = required;
+			this.defaultValue = defaultValue;
+		}
+	}
+
+}

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

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2002-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.handler.annotation.support.reactive;
+
+import java.util.Map;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.handler.annotation.DestinationVariable;
+import org.springframework.messaging.handler.annotation.ValueConstants;
+import org.springframework.util.Assert;
+
+/**
+ * Resolve for {@link DestinationVariable @DestinationVariable} method parameters.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.2
+ */
+public class DestinationVariableMethodArgumentResolver extends AbstractNamedValueMethodArgumentResolver {
+
+	/** The name of the header used to for template variables. */
+	public static final String DESTINATION_TEMPLATE_VARIABLES_HEADER =
+			DestinationVariableMethodArgumentResolver.class.getSimpleName() + ".templateVariables";
+
+
+	public DestinationVariableMethodArgumentResolver(ConversionService conversionService) {
+		super(conversionService, null);
+	}
+
+
+	@Override
+	public boolean supportsParameter(MethodParameter parameter) {
+		return parameter.hasParameterAnnotation(DestinationVariable.class);
+	}
+
+	@Override
+	protected NamedValueInfo createNamedValueInfo(MethodParameter parameter) {
+		DestinationVariable annot = parameter.getParameterAnnotation(DestinationVariable.class);
+		Assert.state(annot != null, "No DestinationVariable annotation");
+		return new DestinationVariableNamedValueInfo(annot);
+	}
+
+	@Override
+	@Nullable
+	@SuppressWarnings("unchecked")
+	protected Object resolveArgumentInternal(MethodParameter parameter, Message<?> message, String name) {
+		MessageHeaders headers = message.getHeaders();
+		Map<String, String> vars = (Map<String, String>) headers.get(DESTINATION_TEMPLATE_VARIABLES_HEADER);
+		return vars != null ? vars.get(name) : null;
+	}
+
+	@Override
+	protected void handleMissingValue(String name, MethodParameter parameter, Message<?> message) {
+		throw new MessageHandlingException(message, "Missing path template variable '" + name + "' " +
+				"for method parameter type [" + parameter.getParameterType() + "]");
+	}
+
+
+	private static final class DestinationVariableNamedValueInfo extends NamedValueInfo {
+
+		private DestinationVariableNamedValueInfo(DestinationVariable annotation) {
+			super(annotation.value(), true, ValueConstants.DEFAULT_NONE);
+		}
+	}
+
+}

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

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2002-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.handler.annotation.support.reactive;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.core.MethodParameter;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.support.NativeMessageHeaderAccessor;
+import org.springframework.util.Assert;
+
+/**
+ * Resolver for {@link Header @Header} arguments. Headers are resolved from
+ * either the top-level header map or the nested
+ * {@link NativeMessageHeaderAccessor native} header map.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.2
+ *
+ * @see HeadersMethodArgumentResolver
+ * @see NativeMessageHeaderAccessor
+ */
+public class HeaderMethodArgumentResolver extends AbstractNamedValueMethodArgumentResolver {
+
+	private static final Log logger = LogFactory.getLog(HeaderMethodArgumentResolver.class);
+
+
+	public HeaderMethodArgumentResolver(
+			ConversionService conversionService, @Nullable ConfigurableBeanFactory beanFactory) {
+
+		super(conversionService, beanFactory);
+	}
+
+
+	@Override
+	public boolean supportsParameter(MethodParameter parameter) {
+		return parameter.hasParameterAnnotation(Header.class);
+	}
+
+	@Override
+	protected NamedValueInfo createNamedValueInfo(MethodParameter parameter) {
+		Header annot = parameter.getParameterAnnotation(Header.class);
+		Assert.state(annot != null, "No Header annotation");
+		return new HeaderNamedValueInfo(annot);
+	}
+
+	@Override
+	@Nullable
+	protected Object resolveArgumentInternal(MethodParameter parameter, Message<?> message, String name) {
+
+		Object headerValue = message.getHeaders().get(name);
+		Object nativeHeaderValue = getNativeHeaderValue(message, name);
+
+		if (headerValue != null && nativeHeaderValue != null) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("A value was found for '" + name + "', in both the top level header map " +
+						"and also in the nested map for native headers. Using the value from top level map. " +
+						"Use 'nativeHeader.myHeader' to resolve the native header.");
+			}
+		}
+
+		return (headerValue != null ? headerValue : nativeHeaderValue);
+	}
+
+	@Nullable
+	private Object getNativeHeaderValue(Message<?> message, String name) {
+		Map<String, List<String>> nativeHeaders = getNativeHeaders(message);
+		if (name.startsWith("nativeHeaders.")) {
+			name = name.substring("nativeHeaders.".length());
+		}
+		if (nativeHeaders == null || !nativeHeaders.containsKey(name)) {
+			return null;
+		}
+		List<?> nativeHeaderValues = nativeHeaders.get(name);
+		return (nativeHeaderValues.size() == 1 ? nativeHeaderValues.get(0) : nativeHeaderValues);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Nullable
+	private Map<String, List<String>> getNativeHeaders(Message<?> message) {
+		return (Map<String, List<String>>) message.getHeaders().get(NativeMessageHeaderAccessor.NATIVE_HEADERS);
+	}
+
+	@Override
+	protected void handleMissingValue(String headerName, MethodParameter parameter, Message<?> message) {
+		throw new MessageHandlingException(message, "Missing header '" + headerName +
+				"' for method parameter type [" + parameter.getParameterType() + "]");
+	}
+
+
+	private static final class HeaderNamedValueInfo extends NamedValueInfo {
+
+		private HeaderNamedValueInfo(Header annotation) {
+			super(annotation.name(), annotation.required(), annotation.defaultValue());
+		}
+	}
+
+}

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

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2002-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.handler.annotation.support.reactive;
+
+import java.lang.reflect.Method;
+import java.util.Map;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.handler.annotation.Headers;
+import org.springframework.messaging.handler.invocation.reactive.SyncHandlerMethodArgumentResolver;
+import org.springframework.messaging.support.MessageHeaderAccessor;
+import org.springframework.util.ReflectionUtils;
+
+/**
+ * Argument resolver for headers. Resolves the following method parameters:
+ * <ul>
+ * <li>{@link Headers @Headers} {@link Map}
+ * <li>{@link MessageHeaders}
+ * <li>{@link MessageHeaderAccessor}
+ * </ul>
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.2
+ */
+public class HeadersMethodArgumentResolver implements SyncHandlerMethodArgumentResolver {
+
+	@Override
+	public boolean supportsParameter(MethodParameter parameter) {
+		Class<?> paramType = parameter.getParameterType();
+		return ((parameter.hasParameterAnnotation(Headers.class) && Map.class.isAssignableFrom(paramType)) ||
+				MessageHeaders.class == paramType || MessageHeaderAccessor.class.isAssignableFrom(paramType));
+	}
+
+	@Override
+	@Nullable
+	public Object resolveArgumentValue(MethodParameter parameter, Message<?> message) {
+		Class<?> paramType = parameter.getParameterType();
+		if (Map.class.isAssignableFrom(paramType)) {
+			return message.getHeaders();
+		}
+		else if (MessageHeaderAccessor.class == paramType) {
+			MessageHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, MessageHeaderAccessor.class);
+			return accessor != null ? accessor : new MessageHeaderAccessor(message);
+		}
+		else if (MessageHeaderAccessor.class.isAssignableFrom(paramType)) {
+			MessageHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, MessageHeaderAccessor.class);
+			if (accessor != null && paramType.isAssignableFrom(accessor.getClass())) {
+				return accessor;
+			}
+			else {
+				Method method = ReflectionUtils.findMethod(paramType, "wrap", Message.class);
+				if (method == null) {
+					throw new IllegalStateException(
+							"Cannot create accessor of type " + paramType + " for message " + message);
+				}
+				return ReflectionUtils.invokeMethod(method, null, message);
+			}
+		}
+		else {
+			throw new IllegalStateException("Unexpected parameter of type " + paramType +
+					" in method " + parameter.getMethod() + ". ");
+		}
+	}
+
+}

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

@@ -24,9 +24,14 @@ import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 
+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.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.handler.CompositeMessageCondition;
@@ -73,6 +78,8 @@ public class MessageMappingMessageHandler extends AbstractMethodMessageHandler<C
 	@Nullable
 	private HandlerMethodReturnValueHandler encoderReturnValueHandler;
 
+	private ConversionService conversionService = new DefaultFormattingConversionService();
+
 	@Nullable
 	private StringValueResolver valueResolver;
 
@@ -149,6 +156,23 @@ public class MessageMappingMessageHandler extends AbstractMethodMessageHandler<C
 		return this.encoderReturnValueHandler;
 	}
 
+	/**
+	 * Configure a {@link ConversionService} to use for type conversion of
+	 * String based values, e.g. in destination variables or headers.
+	 * <p>By default {@link DefaultFormattingConversionService} is used.
+	 * @param conversionService the conversion service to use
+	 */
+	public void setConversionService(ConversionService conversionService) {
+		this.conversionService = conversionService;
+	}
+
+	/**
+	 * Return the configured ConversionService.
+	 */
+	public ConversionService getConversionService() {
+		return this.conversionService;
+	}
+
 	@Override
 	public void setEmbeddedValueResolver(StringValueResolver resolver) {
 		this.valueResolver = resolver;
@@ -159,6 +183,15 @@ public class MessageMappingMessageHandler extends AbstractMethodMessageHandler<C
 	protected List<? extends HandlerMethodArgumentResolver> initArgumentResolvers() {
 		List<HandlerMethodArgumentResolver> resolvers = new ArrayList<>();
 
+		ApplicationContext context = getApplicationContext();
+		ConfigurableBeanFactory beanFactory = (context instanceof ConfigurableApplicationContext ?
+				((ConfigurableApplicationContext) context).getBeanFactory() : null);
+
+		// Annotation-based resolvers
+		resolvers.add(new HeaderMethodArgumentResolver(this.conversionService, beanFactory));
+		resolvers.add(new HeadersMethodArgumentResolver());
+		resolvers.add(new DestinationVariableMethodArgumentResolver(this.conversionService));
+
 		// Custom resolvers
 		resolvers.addAll(getArgumentResolverConfigurer().getCustomResolvers());
 

spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/reactive/SyncHandlerMethodArgumentResolver.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2002-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.messaging.handler.invocation.reactive;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+
+/**
+ * An extension of {@link HandlerMethodArgumentResolver} for implementations
+ * that are synchronous in nature and do not block to resolve values.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.2
+ */
+public interface SyncHandlerMethodArgumentResolver extends HandlerMethodArgumentResolver {
+
+	/**
+	 * {@inheritDoc}
+	 * <p>By default this simply delegates to {@link #resolveArgumentValue} for
+	 * synchronous resolution.
+	 */
+	@Override
+	default Mono<Object> resolveArgument(MethodParameter parameter, Message<?> message) {
+		return Mono.justOrEmpty(resolveArgumentValue(parameter, message));
+	}
+
+	/**
+	 * Resolve the value for the method parameter synchronously.
+	 * @param parameter the method parameter
+	 * @param message the currently processed message
+	 * @return the resolved value, if any
+	 */
+	@Nullable
+	Object resolveArgumentValue(MethodParameter parameter, Message<?> message);
+
+}

spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/MessagingPredicates.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2002-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.messaging.handler.annotation;
+
+import java.util.function.Predicate;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.lang.Nullable;
+
+/**
+ * Predicates for messaging annotations.
+ *
+ * @author Rossen Stoyanchev
+ */
+public class MessagingPredicates {
+
+	public static DestinationVariablePredicate destinationVar() {
+		return new DestinationVariablePredicate();
+	}
+
+	public static DestinationVariablePredicate destinationVar(String value) {
+		return new DestinationVariablePredicate().value(value);
+	}
+
+	public static HeaderPredicate header() {
+		return new HeaderPredicate();
+	}
+
+	public static HeaderPredicate header(String name) {
+		return new HeaderPredicate().name(name);
+	}
+
+	public static HeaderPredicate header(String name, String defaultValue) {
+		return new HeaderPredicate().name(name).defaultValue(defaultValue);
+	}
+
+	public static HeaderPredicate headerPlain() {
+		return new HeaderPredicate().noAttributes();
+	}
+
+
+	public static class DestinationVariablePredicate implements Predicate<MethodParameter> {
+
+		@Nullable
+		private String value;
+
+
+		public DestinationVariablePredicate value(@Nullable String name) {
+			this.value = name;
+			return this;
+		}
+
+		public DestinationVariablePredicate noValue() {
+			this.value = "";
+			return this;
+		}
+
+		@Override
+		public boolean test(MethodParameter parameter) {
+			DestinationVariable annotation = parameter.getParameterAnnotation(DestinationVariable.class);
+			return annotation != null && (this.value == null || annotation.value().equals(this.value));
+		}
+	}
+
+
+	public static class HeaderPredicate implements Predicate<MethodParameter> {
+
+		@Nullable
+		private String name;
+
+		@Nullable
+		private Boolean required;
+
+		@Nullable
+		private String defaultValue;
+
+
+		public HeaderPredicate name(@Nullable String name) {
+			this.name = name;
+			return this;
+		}
+
+		public HeaderPredicate noName() {
+			this.name = "";
+			return this;
+		}
+
+		public HeaderPredicate required(boolean required) {
+			this.required = required;
+			return this;
+		}
+
+		public HeaderPredicate defaultValue(@Nullable String value) {
+			this.defaultValue = value;
+			return this;
+		}
+
+		public HeaderPredicate noAttributes() {
+			this.name = "";
+			this.required = true;
+			this.defaultValue = ValueConstants.DEFAULT_NONE;
+			return this;
+		}
+
+		@Override
+		public boolean test(MethodParameter parameter) {
+			Header annotation = parameter.getParameterAnnotation(Header.class);
+			return annotation != null &&
+					(this.name == null || annotation.name().equals(this.name)) &&
+					(this.required == null || annotation.required() == this.required) &&
+					(this.defaultValue == null || annotation.defaultValue().equals(this.defaultValue));
+		}
+	}
+
+}

spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/support/reactive/DestinationVariableMethodArgumentResolverTests.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2002-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.handler.annotation.support.reactive;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.core.convert.support.DefaultConversionService;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.handler.annotation.DestinationVariable;
+import org.springframework.messaging.handler.invocation.ResolvableMethod;
+import org.springframework.messaging.support.MessageBuilder;
+
+import static org.junit.Assert.*;
+import static org.springframework.messaging.handler.annotation.MessagingPredicates.*;
+
+/**
+ * Test fixture for {@link DestinationVariableMethodArgumentResolver} tests.
+ * @author Rossen Stoyanchev
+ */
+public class DestinationVariableMethodArgumentResolverTests {
+
+	private final DestinationVariableMethodArgumentResolver resolver =
+			new DestinationVariableMethodArgumentResolver(new DefaultConversionService());
+
+	private final ResolvableMethod resolvable =
+			ResolvableMethod.on(getClass()).named("handleMessage").build();
+
+
+	@Test
+	public void supportsParameter() {
+		assertTrue(resolver.supportsParameter(this.resolvable.annot(destinationVar().noValue()).arg()));
+		assertFalse(resolver.supportsParameter(this.resolvable.annotNotPresent(DestinationVariable.class).arg()));
+	}
+
+	@Test
+	public void resolveArgument() {
+
+		Map<String, Object> vars = new HashMap<>();
+		vars.put("foo", "bar");
+		vars.put("name", "value");
+
+		Message<byte[]> message = MessageBuilder.withPayload(new byte[0]).setHeader(
+			DestinationVariableMethodArgumentResolver.DESTINATION_TEMPLATE_VARIABLES_HEADER, vars).build();
+
+		Object result = resolveArgument(this.resolvable.annot(destinationVar().noValue()).arg(), message);
+		assertEquals("bar", result);
+
+		result = resolveArgument(this.resolvable.annot(destinationVar("name")).arg(), message);
+		assertEquals("value", result);
+	}
+
+	@Test(expected = MessageHandlingException.class)
+	public void resolveArgumentNotFound() {
+		Message<byte[]> message = MessageBuilder.withPayload(new byte[0]).build();
+		resolveArgument(this.resolvable.annot(destinationVar().noValue()).arg(), message);
+	}
+
+	@SuppressWarnings({"unchecked", "ConstantConditions"})
+	private <T> T resolveArgument(MethodParameter param, Message<?> message) {
+		return (T) this.resolver.resolveArgument(param, message).block(Duration.ofSeconds(5));
+	}
+
+
+	@SuppressWarnings("unused")
+	private void handleMessage(
+			@DestinationVariable String foo,
+			@DestinationVariable(value = "name") String param1,
+			String param3) {
+	}
+
+}

spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/support/reactive/HeaderMethodArgumentResolverTests.java

@@ -0,0 +1,176 @@
+/*
+ * Copyright 2002-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.handler.annotation.support.reactive;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.core.MethodParameter;
+import org.springframework.core.convert.support.DefaultConversionService;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.handler.invocation.ResolvableMethod;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.messaging.support.NativeMessageHeaderAccessor;
+
+import static org.junit.Assert.*;
+import static org.springframework.messaging.handler.annotation.MessagingPredicates.*;
+
+/**
+ * Test fixture for {@link HeaderMethodArgumentResolver} tests.
+ * @author Rossen Stoyanchev
+ */
+public class HeaderMethodArgumentResolverTests {
+
+	private HeaderMethodArgumentResolver resolver;
+
+	private final ResolvableMethod resolvable = ResolvableMethod.on(getClass()).named("handleMessage").build();
+
+
+	@Before
+	public void setup() {
+		GenericApplicationContext context = new GenericApplicationContext();
+		context.refresh();
+		this.resolver = new HeaderMethodArgumentResolver(new DefaultConversionService(), context.getBeanFactory());
+	}
+
+
+	@Test
+	public void supportsParameter() {
+		assertTrue(this.resolver.supportsParameter(this.resolvable.annot(headerPlain()).arg()));
+		assertFalse(this.resolver.supportsParameter(this.resolvable.annotNotPresent(Header.class).arg()));
+	}
+
+	@Test
+	public void resolveArgument() {
+		Message<byte[]> message = MessageBuilder.withPayload(new byte[0]).setHeader("param1", "foo").build();
+		Object result = resolveArgument(this.resolvable.annot(headerPlain()).arg(), message);
+		assertEquals("foo", result);
+	}
+
+	@Test  // SPR-11326
+	public void resolveArgumentNativeHeader() {
+		TestMessageHeaderAccessor headers = new TestMessageHeaderAccessor();
+		headers.setNativeHeader("param1", "foo");
+		Message<byte[]> message = MessageBuilder.withPayload(new byte[0]).setHeaders(headers).build();
+		assertEquals("foo", resolveArgument(this.resolvable.annot(headerPlain()).arg(), message));
+	}
+
+	@Test
+	public void resolveArgumentNativeHeaderAmbiguity() {
+		TestMessageHeaderAccessor headers = new TestMessageHeaderAccessor();
+		headers.setHeader("param1", "foo");
+		headers.setNativeHeader("param1", "native-foo");
+		Message<byte[]> message = MessageBuilder.withPayload(new byte[0]).setHeaders(headers).build();
+
+		assertEquals("foo", resolveArgument(
+				this.resolvable.annot(headerPlain()).arg(), message));
+
+		assertEquals("native-foo", resolveArgument(
+				this.resolvable.annot(header("nativeHeaders.param1")).arg(), message));
+	}
+
+	@Test(expected = MessageHandlingException.class)
+	public void resolveArgumentNotFound() {
+		Message<byte[]> message = MessageBuilder.withPayload(new byte[0]).build();
+		resolveArgument(this.resolvable.annot(headerPlain()).arg(), message);
+	}
+
+	@Test
+	public void resolveArgumentDefaultValue() {
+		Message<byte[]> message = MessageBuilder.withPayload(new byte[0]).build();
+		Object result = resolveArgument(this.resolvable.annot(header("name", "bar")).arg(), message);
+		assertEquals("bar", result);
+	}
+
+	@Test
+	public void resolveDefaultValueSystemProperty() {
+		System.setProperty("systemProperty", "sysbar");
+		try {
+			Message<byte[]> message = MessageBuilder.withPayload(new byte[0]).build();
+			MethodParameter param = this.resolvable.annot(header("name", "#{systemProperties.systemProperty}")).arg();
+			Object result = resolveArgument(param, message);
+			assertEquals("sysbar", result);
+		}
+		finally {
+			System.clearProperty("systemProperty");
+		}
+	}
+
+	@Test
+	public void resolveNameFromSystemProperty() {
+		System.setProperty("systemProperty", "sysbar");
+		try {
+			Message<byte[]> message = MessageBuilder.withPayload(new byte[0]).setHeader("sysbar", "foo").build();
+			MethodParameter param = this.resolvable.annot(header("#{systemProperties.systemProperty}")).arg();
+			Object result = resolveArgument(param, message);
+			assertEquals("foo", result);
+		}
+		finally {
+			System.clearProperty("systemProperty");
+		}
+	}
+
+	@Test
+	public void resolveOptionalHeaderWithValue() {
+		Message<String> message = MessageBuilder.withPayload("foo").setHeader("foo", "bar").build();
+		MethodParameter param = this.resolvable.annot(header("foo")).arg(Optional.class, String.class);
+		Object result = resolveArgument(param, message);
+		assertEquals(Optional.of("bar"), result);
+	}
+
+	@Test
+	public void resolveOptionalHeaderAsEmpty() {
+		Message<String> message = MessageBuilder.withPayload("foo").build();
+		MethodParameter param = this.resolvable.annot(header("foo")).arg(Optional.class, String.class);
+		Object result = resolveArgument(param, message);
+		assertEquals(Optional.empty(), result);
+	}
+
+	@SuppressWarnings({"unchecked", "ConstantConditions"})
+	private <T> T resolveArgument(MethodParameter param, Message<?> message) {
+		return (T) this.resolver.resolveArgument(param, message).block(Duration.ofSeconds(5));
+	}
+
+
+	@SuppressWarnings({"unused", "OptionalUsedAsFieldOrParameterType"})
+	public void handleMessage(
+			@Header String param1,
+			@Header(name = "name", defaultValue = "bar") String param2,
+			@Header(name = "name", defaultValue = "#{systemProperties.systemProperty}") String param3,
+			@Header(name = "#{systemProperties.systemProperty}") String param4,
+			String param5,
+			@Header("foo") Optional<String> param6,
+			@Header("nativeHeaders.param1") String nativeHeaderParam1) {
+	}
+
+
+	public static class TestMessageHeaderAccessor extends NativeMessageHeaderAccessor {
+
+		TestMessageHeaderAccessor() {
+			super((Map<String, List<String>>) null);
+		}
+	}
+
+}

spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/support/reactive/HeadersMethodArgumentResolverTests.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2002-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.handler.annotation.support.reactive;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Map;
+
+import org.junit.Test;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.handler.annotation.Headers;
+import org.springframework.messaging.handler.invocation.ResolvableMethod;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.messaging.support.MessageHeaderAccessor;
+import org.springframework.messaging.support.NativeMessageHeaderAccessor;
+
+import static org.junit.Assert.*;
+
+/**
+ * Test fixture for {@link HeadersMethodArgumentResolver} tests.
+ * @author Rossen Stoyanchev
+ */
+public class HeadersMethodArgumentResolverTests {
+
+	private final HeadersMethodArgumentResolver resolver = new HeadersMethodArgumentResolver();
+
+	private Message<byte[]> message =
+			MessageBuilder.withPayload(new byte[0]).copyHeaders(Collections.singletonMap("foo", "bar")).build();
+
+	private final ResolvableMethod resolvable = ResolvableMethod.on(getClass()).named("handleMessage").build();
+
+
+	@Test
+	public void supportsParameter() {
+
+		assertTrue(this.resolver.supportsParameter(
+				this.resolvable.annotPresent(Headers.class).arg(Map.class, String.class, Object.class)));
+
+		assertTrue(this.resolver.supportsParameter(this.resolvable.arg(MessageHeaders.class)));
+		assertTrue(this.resolver.supportsParameter(this.resolvable.arg(MessageHeaderAccessor.class)));
+		assertTrue(this.resolver.supportsParameter(this.resolvable.arg(TestMessageHeaderAccessor.class)));
+
+		assertFalse(this.resolver.supportsParameter(this.resolvable.annotPresent(Headers.class).arg(String.class)));
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void resolveArgumentAnnotated() {
+		MethodParameter param = this.resolvable.annotPresent(Headers.class).arg(Map.class, String.class, Object.class);
+		Map<String, Object> headers = resolveArgument(param);
+		assertEquals("bar", headers.get("foo"));
+	}
+
+	@Test(expected = IllegalStateException.class)
+	public void resolveArgumentAnnotatedNotMap() {
+		resolveArgument(this.resolvable.annotPresent(Headers.class).arg(String.class));
+	}
+
+	@Test
+	public void resolveArgumentMessageHeaders() {
+		MessageHeaders headers = resolveArgument(this.resolvable.arg(MessageHeaders.class));
+		assertEquals("bar", headers.get("foo"));
+	}
+
+	@Test
+	public void resolveArgumentMessageHeaderAccessor() {
+		MessageHeaderAccessor headers = resolveArgument(this.resolvable.arg(MessageHeaderAccessor.class));
+		assertEquals("bar", headers.getHeader("foo"));
+	}
+
+	@Test
+	public void resolveArgumentMessageHeaderAccessorSubclass() {
+		TestMessageHeaderAccessor headers = resolveArgument(this.resolvable.arg(TestMessageHeaderAccessor.class));
+		assertEquals("bar", headers.getHeader("foo"));
+	}
+
+	@SuppressWarnings({"unchecked", "ConstantConditions"})
+	private <T> T resolveArgument(MethodParameter param) {
+		return (T) this.resolver.resolveArgument(param, this.message).block(Duration.ofSeconds(5));
+	}
+
+
+	@SuppressWarnings("unused")
+	private void handleMessage(
+			@Headers Map<String, Object> param1,
+			@Headers String param2,
+			MessageHeaders param3,
+			MessageHeaderAccessor param4,
+			TestMessageHeaderAccessor param5) {
+	}
+
+
+	public static class TestMessageHeaderAccessor extends NativeMessageHeaderAccessor {
+
+		TestMessageHeaderAccessor(Message<?> message) {
+			super(message);
+		}
+
+		public static TestMessageHeaderAccessor wrap(Message<?> message) {
+			return new TestMessageHeaderAccessor(message);
+		}
+	}
+
+}