Reactive InvocableHandlerMethod in spring-messaging

See gh-21987

spring-messaging/spring-messaging.gradle

@@ -24,6 +24,7 @@ dependencies {
 		exclude group: "org.springframework", module: "spring-context"
 	}
 	testCompile("org.apache.activemq:activemq-stomp:5.8.0")
+	testCompile("io.projectreactor:reactor-test")
 	testCompile("org.jetbrains.kotlin:kotlin-reflect:${kotlinVersion}")
 	testCompile("org.jetbrains.kotlin:kotlin-stdlib:${kotlinVersion}")
 	testCompile("org.xmlunit:xmlunit-matchers:2.6.2")

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

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2002-2018 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.messaging.Message;
+
+/**
+ * Strategy interface for resolving method parameters into argument values
+ * in the context of a given {@link Message}.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.2
+ */
+public interface HandlerMethodArgumentResolver {
+
+	/**
+	 * Whether the given {@linkplain MethodParameter method parameter} is
+	 * supported by this resolver.
+	 * @param parameter the method parameter to check
+	 * @return {@code true} if this resolver supports the supplied parameter;
+	 * {@code false} otherwise
+	 */
+	boolean supportsParameter(MethodParameter parameter);
+
+	/**
+	 * Resolves a method parameter into an argument value from a given message.
+	 * @param parameter the method parameter to resolve.
+	 * This parameter must have previously been passed to
+	 * {@link #supportsParameter(org.springframework.core.MethodParameter)}
+	 * which must have returned {@code true}.
+	 * @param message the currently processed message
+	 * @return {@code Mono} for the argument value, possibly empty
+	 */
+	Mono<Object> resolveArgument(MethodParameter parameter, Message<?> message);
+
+}

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

@@ -0,0 +1,142 @@
+/*
+ * Copyright 2002-2018 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 java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+
+/**
+ * Resolves method parameters by delegating to a list of registered
+ * {@link HandlerMethodArgumentResolver HandlerMethodArgumentResolvers}.
+ * Previously resolved method parameters are cached for faster lookups.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.2
+ */
+class HandlerMethodArgumentResolverComposite implements HandlerMethodArgumentResolver {
+
+	protected final Log logger = LogFactory.getLog(getClass());
+
+	private final List<HandlerMethodArgumentResolver> argumentResolvers = new LinkedList<>();
+
+	private final Map<MethodParameter, HandlerMethodArgumentResolver> argumentResolverCache =
+			new ConcurrentHashMap<>(256);
+
+
+	/**
+	 * Add the given {@link HandlerMethodArgumentResolver}.
+	 */
+	public HandlerMethodArgumentResolverComposite addResolver(HandlerMethodArgumentResolver resolver) {
+		this.argumentResolvers.add(resolver);
+		return this;
+	}
+
+	/**
+	 * Add the given {@link HandlerMethodArgumentResolver HandlerMethodArgumentResolvers}.
+	 */
+	public HandlerMethodArgumentResolverComposite addResolvers(@Nullable HandlerMethodArgumentResolver... resolvers) {
+		if (resolvers != null) {
+			Collections.addAll(this.argumentResolvers, resolvers);
+		}
+		return this;
+	}
+
+	/**
+	 * Add the given {@link HandlerMethodArgumentResolver HandlerMethodArgumentResolvers}.
+	 */
+	public HandlerMethodArgumentResolverComposite addResolvers(
+			@Nullable List<? extends HandlerMethodArgumentResolver> resolvers) {
+
+		if (resolvers != null) {
+			this.argumentResolvers.addAll(resolvers);
+		}
+		return this;
+	}
+
+	/**
+	 * Return a read-only list with the contained resolvers, or an empty list.
+	 */
+	public List<HandlerMethodArgumentResolver> getResolvers() {
+		return Collections.unmodifiableList(this.argumentResolvers);
+	}
+
+	/**
+	 * Clear the list of configured resolvers.
+	 */
+	public void clear() {
+		this.argumentResolvers.clear();
+	}
+
+
+	/**
+	 * Whether the given {@linkplain MethodParameter method parameter} is
+	 * supported by any registered {@link HandlerMethodArgumentResolver}.
+	 */
+	@Override
+	public boolean supportsParameter(MethodParameter parameter) {
+		return getArgumentResolver(parameter) != null;
+	}
+
+	/**
+	 * Iterate over registered
+	 * {@link HandlerMethodArgumentResolver HandlerMethodArgumentResolvers} and
+	 * invoke the one that supports it.
+	 * @throws IllegalStateException if no suitable
+	 * {@link HandlerMethodArgumentResolver} is found.
+	 */
+	@Override
+	public Mono<Object> resolveArgument(MethodParameter parameter, Message<?> message) {
+		HandlerMethodArgumentResolver resolver = getArgumentResolver(parameter);
+		if (resolver == null) {
+			throw new IllegalArgumentException(
+					"Unsupported parameter type [" + parameter.getParameterType().getName() + "]." +
+							" supportsParameter should be called first.");
+		}
+		return resolver.resolveArgument(parameter, message);
+	}
+
+	/**
+	 * Find a registered {@link HandlerMethodArgumentResolver} that supports
+	 * the given method parameter.
+	 */
+	@Nullable
+	private HandlerMethodArgumentResolver getArgumentResolver(MethodParameter parameter) {
+		HandlerMethodArgumentResolver result = this.argumentResolverCache.get(parameter);
+		if (result == null) {
+			for (HandlerMethodArgumentResolver methodArgumentResolver : this.argumentResolvers) {
+				if (methodArgumentResolver.supportsParameter(parameter)) {
+					result = methodArgumentResolver;
+					this.argumentResolverCache.put(parameter, result);
+					break;
+				}
+			}
+		}
+		return result;
+	}
+
+}

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

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2002-2017 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;
+
+/**
+ * Handle the return value from the invocation of an annotated {@link Message}
+ * handling method.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.2
+ */
+public interface HandlerMethodReturnValueHandler {
+
+	/**
+	 * Whether the given {@linkplain MethodParameter method return type} is
+	 * supported by this handler.
+	 * @param returnType the method return type to check
+	 * @return {@code true} if this handler supports the supplied return type;
+	 * {@code false} otherwise
+	 */
+	boolean supportsReturnType(MethodParameter returnType);
+
+	/**
+	 * Handle the given return value.
+	 * @param returnValue the value returned from the handler method
+	 * @param returnType the type of the return value. This type must have previously
+	 * been passed to {@link #supportsReturnType(MethodParameter)}
+	 * and it must have returned {@code true}.
+	 * @return {@code Mono<Void>} to indicate when handling is complete.
+	 */
+	Mono<Void> handleReturnValue(@Nullable Object returnValue, MethodParameter returnType, Message<?> message);
+
+}

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

@@ -0,0 +1,108 @@
+/*
+ * Copyright 2002-2018 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 java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+
+/**
+ * A HandlerMethodReturnValueHandler that wraps and delegates to others.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.2
+ */
+public class HandlerMethodReturnValueHandlerComposite implements HandlerMethodReturnValueHandler {
+
+	protected final Log logger = LogFactory.getLog(getClass());
+
+
+	private final List<HandlerMethodReturnValueHandler> returnValueHandlers = new ArrayList<>();
+
+
+	/**
+	 * Return a read-only list with the configured handlers.
+	 */
+	public List<HandlerMethodReturnValueHandler> getReturnValueHandlers() {
+		return Collections.unmodifiableList(this.returnValueHandlers);
+	}
+
+	/**
+	 * Clear the list of configured handlers.
+	 */
+	public void clear() {
+		this.returnValueHandlers.clear();
+	}
+
+	/**
+	 * Add the given {@link HandlerMethodReturnValueHandler}.
+	 */
+	public HandlerMethodReturnValueHandlerComposite addHandler(HandlerMethodReturnValueHandler returnValueHandler) {
+		this.returnValueHandlers.add(returnValueHandler);
+		return this;
+	}
+
+	/**
+	 * Add the given {@link HandlerMethodReturnValueHandler HandlerMethodReturnValueHandlers}.
+	 */
+	public HandlerMethodReturnValueHandlerComposite addHandlers(
+			@Nullable List<? extends HandlerMethodReturnValueHandler> handlers) {
+
+		if (handlers != null) {
+			this.returnValueHandlers.addAll(handlers);
+		}
+		return this;
+	}
+
+	@Override
+	public boolean supportsReturnType(MethodParameter returnType) {
+		return getReturnValueHandler(returnType) != null;
+	}
+
+	@Override
+	public Mono<Void> handleReturnValue(@Nullable Object returnValue, MethodParameter returnType, Message<?> message) {
+		HandlerMethodReturnValueHandler handler = getReturnValueHandler(returnType);
+		if (handler == null) {
+			throw new IllegalStateException("No handler for return value type: " + returnType.getParameterType());
+		}
+		if (logger.isTraceEnabled()) {
+			logger.trace("Processing return value with " + handler);
+		}
+		return handler.handleReturnValue(returnValue, returnType, message);
+	}
+
+	@SuppressWarnings("ForLoopReplaceableByForEach")
+	@Nullable
+	private HandlerMethodReturnValueHandler getReturnValueHandler(MethodParameter returnType) {
+		for (int i = 0; i < this.returnValueHandlers.size(); i++) {
+			HandlerMethodReturnValueHandler handler = this.returnValueHandlers.get(i);
+			if (handler.supportsReturnType(returnType)) {
+				return handler;
+			}
+		}
+		return null;
+	}
+
+}

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

@@ -0,0 +1,213 @@
+/*
+ * Copyright 2002-2018 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 java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Stream;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.DefaultParameterNameDiscoverer;
+import org.springframework.core.MethodParameter;
+import org.springframework.core.ParameterNameDiscoverer;
+import org.springframework.core.ReactiveAdapter;
+import org.springframework.core.ReactiveAdapterRegistry;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.handler.HandlerMethod;
+import org.springframework.messaging.handler.invocation.MethodArgumentResolutionException;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.ReflectionUtils;
+
+/**
+ * Extension of {@link HandlerMethod} that invokes the underlying method with
+ * argument values resolved from the current HTTP request through a list of
+ * {@link HandlerMethodArgumentResolver}.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.2
+ */
+public class InvocableHandlerMethod extends HandlerMethod {
+
+	private static final Mono<Object[]> EMPTY_ARGS = Mono.just(new Object[0]);
+
+	private static final Object NO_ARG_VALUE = new Object();
+
+
+	private HandlerMethodArgumentResolverComposite resolvers = new HandlerMethodArgumentResolverComposite();
+
+	private ParameterNameDiscoverer parameterNameDiscoverer = new DefaultParameterNameDiscoverer();
+
+	private ReactiveAdapterRegistry reactiveAdapterRegistry = ReactiveAdapterRegistry.getSharedInstance();
+
+
+	/**
+	 * Create an instance from a {@code HandlerMethod}.
+	 */
+	public InvocableHandlerMethod(HandlerMethod handlerMethod) {
+		super(handlerMethod);
+	}
+
+	/**
+	 * Create an instance from a bean instance and a method.
+	 */
+	public InvocableHandlerMethod(Object bean, Method method) {
+		super(bean, method);
+	}
+
+
+	/**
+	 * Configure the argument resolvers to use to use for resolving method
+	 * argument values against a {@code ServerWebExchange}.
+	 */
+	public void setArgumentResolvers(List<HandlerMethodArgumentResolver> resolvers) {
+		this.resolvers.addResolvers(resolvers);
+	}
+
+	/**
+	 * Return the configured argument resolvers.
+	 */
+	public List<HandlerMethodArgumentResolver> getResolvers() {
+		return this.resolvers.getResolvers();
+	}
+
+	/**
+	 * Set the ParameterNameDiscoverer for resolving parameter names when needed
+	 * (e.g. default request attribute name).
+	 * <p>Default is a {@link DefaultParameterNameDiscoverer}.
+	 */
+	public void setParameterNameDiscoverer(ParameterNameDiscoverer nameDiscoverer) {
+		this.parameterNameDiscoverer = nameDiscoverer;
+	}
+
+	/**
+	 * Return the configured parameter name discoverer.
+	 */
+	public ParameterNameDiscoverer getParameterNameDiscoverer() {
+		return this.parameterNameDiscoverer;
+	}
+
+	/**
+	 * Configure a reactive registry. This is needed for cases where the response
+	 * is fully handled within the controller in combination with an async void
+	 * return value.
+	 * <p>By default this is an instance of {@link ReactiveAdapterRegistry} with
+	 * default settings.
+	 * @param registry the registry to use
+	 */
+	public void setReactiveAdapterRegistry(ReactiveAdapterRegistry registry) {
+		this.reactiveAdapterRegistry = registry;
+	}
+
+
+	/**
+	 * Invoke the method for the given exchange.
+	 * @param message the current message
+	 * @param providedArgs optional list of argument values to match by type
+	 * @return a Mono with the result from the invocation.
+	 */
+	public Mono<Object> invoke(Message<?> message, Object... providedArgs) {
+
+		return getMethodArgumentValues(message, providedArgs).flatMap(args -> {
+			Object value;
+			try {
+				ReflectionUtils.makeAccessible(getBridgedMethod());
+				value = getBridgedMethod().invoke(getBean(), args);
+			}
+			catch (IllegalArgumentException ex) {
+				assertTargetBean(getBridgedMethod(), getBean(), args);
+				String text = (ex.getMessage() != null ? ex.getMessage() : "Illegal argument");
+				return Mono.error(new IllegalStateException(formatInvokeError(text, args), ex));
+			}
+			catch (InvocationTargetException ex) {
+				return Mono.error(ex.getTargetException());
+			}
+			catch (Throwable ex) {
+				// Unlikely to ever get here, but it must be handled...
+				return Mono.error(new IllegalStateException(formatInvokeError("Invocation failure", args), ex));
+			}
+
+			MethodParameter returnType = getReturnType();
+			ReactiveAdapter adapter = this.reactiveAdapterRegistry.getAdapter(returnType.getParameterType());
+			return isAsyncVoidReturnType(returnType, adapter) ?
+					Mono.from(adapter.toPublisher(value)) : Mono.justOrEmpty(value);
+		});
+	}
+
+	private Mono<Object[]> getMethodArgumentValues(Message<?> message, Object... providedArgs) {
+		if (ObjectUtils.isEmpty(getMethodParameters())) {
+			return EMPTY_ARGS;
+		}
+		MethodParameter[] parameters = getMethodParameters();
+		List<Mono<Object>> argMonos = new ArrayList<>(parameters.length);
+		for (MethodParameter parameter : parameters) {
+			parameter.initParameterNameDiscovery(this.parameterNameDiscoverer);
+			Object providedArg = findProvidedArgument(parameter, providedArgs);
+			if (providedArg != null) {
+				argMonos.add(Mono.just(providedArg));
+				continue;
+			}
+			if (!this.resolvers.supportsParameter(parameter)) {
+				return Mono.error(new MethodArgumentResolutionException(
+						message, parameter, formatArgumentError(parameter, "No suitable resolver")));
+			}
+			try {
+				argMonos.add(this.resolvers.resolveArgument(parameter, message)
+						.defaultIfEmpty(NO_ARG_VALUE)
+						.doOnError(cause -> logArgumentErrorIfNecessary(parameter, cause)));
+			}
+			catch (Exception ex) {
+				logArgumentErrorIfNecessary(parameter, ex);
+				argMonos.add(Mono.error(ex));
+			}
+		}
+		return Mono.zip(argMonos, values ->
+				Stream.of(values).map(o -> o != NO_ARG_VALUE ? o : null).toArray());
+	}
+
+	private void logArgumentErrorIfNecessary(MethodParameter parameter, Throwable cause) {
+		// Leave stack trace for later, if error is not handled..
+		String causeMessage = cause.getMessage();
+		if (!causeMessage.contains(parameter.getExecutable().toGenericString())) {
+			if (logger.isDebugEnabled()) {
+				logger.debug(formatArgumentError(parameter, causeMessage));
+			}
+		}
+	}
+
+	private boolean isAsyncVoidReturnType(MethodParameter returnType, @Nullable ReactiveAdapter reactiveAdapter) {
+		if (reactiveAdapter != null && reactiveAdapter.supportsEmpty()) {
+			if (reactiveAdapter.isNoValue()) {
+				return true;
+			}
+			Type parameterType = returnType.getGenericParameterType();
+			if (parameterType instanceof ParameterizedType) {
+				ParameterizedType type = (ParameterizedType) parameterType;
+				if (type.getActualTypeArguments().length == 1) {
+					return Void.class.equals(type.getActualTypeArguments()[0]);
+				}
+			}
+		}
+		return false;
+	}
+
+}

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

@@ -0,0 +1,10 @@
+/**
+ * Common infrastructure for invoking message handler methods with non-blocking,
+ * and reactive contracts.
+ */
+@NonNullApi
+@NonNullFields
+package org.springframework.messaging.handler.invocation.reactive;
+
+import org.springframework.lang.NonNullApi;
+import org.springframework.lang.NonNullFields;

spring-messaging/src/test/java/org/springframework/messaging/handler/invocation/reactive/InvocableHandlerMethodTests.java

@@ -0,0 +1,237 @@
+/*
+ * Copyright 2002-2018 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 java.lang.reflect.Method;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.Test;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.handler.ResolvableMethod;
+import org.springframework.messaging.handler.invocation.MethodArgumentResolutionException;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+/**
+ * Unit tests for {@link InvocableHandlerMethod}.
+ *
+ * @author Rossen Stoyanchev
+ * @author Juergen Hoeller
+ */
+public class InvocableHandlerMethodTests {
+
+	private final Message<?> message = mock(Message.class);
+
+	private final List<HandlerMethodArgumentResolver> resolvers = new ArrayList<>();
+
+
+	@Test
+	public void resolveArg() {
+		this.resolvers.add(new StubArgumentResolver(99));
+		this.resolvers.add(new StubArgumentResolver("value"));
+		Method method = ResolvableMethod.on(Handler.class).mockCall(c -> c.handle(0, "")).method();
+		Object value = invokeAndBlock(new Handler(), method);
+
+		assertEquals(1, getStubResolver(0).getResolvedParameters().size());
+		assertEquals(1, getStubResolver(1).getResolvedParameters().size());
+		assertEquals("99-value", value);
+		assertEquals("intArg", getStubResolver(0).getResolvedParameters().get(0).getParameterName());
+		assertEquals("stringArg", getStubResolver(1).getResolvedParameters().get(0).getParameterName());
+	}
+
+	@Test
+	public void resolveNoArgValue() {
+		this.resolvers.add(new StubArgumentResolver(Integer.class));
+		this.resolvers.add(new StubArgumentResolver(String.class));
+		Method method = ResolvableMethod.on(Handler.class).mockCall(c -> c.handle(0, "")).method();
+		Object value = invokeAndBlock(new Handler(), method);
+
+		assertEquals(1, getStubResolver(0).getResolvedParameters().size());
+		assertEquals(1, getStubResolver(1).getResolvedParameters().size());
+		assertEquals("null-null", value);
+	}
+
+	@Test
+	public void cannotResolveArg() {
+		try {
+			Method method = ResolvableMethod.on(Handler.class).mockCall(c -> c.handle(0, "")).method();
+			invokeAndBlock(new Handler(), method);
+			fail("Expected exception");
+		}
+		catch (MethodArgumentResolutionException ex) {
+			assertNotNull(ex.getMessage());
+			assertTrue(ex.getMessage().contains("Could not resolve parameter [0]"));
+		}
+	}
+
+	@Test
+	public void resolveProvidedArg() {
+		Method method = ResolvableMethod.on(Handler.class).mockCall(c -> c.handle(0, "")).method();
+		Object value = invokeAndBlock(new Handler(), method, 99, "value");
+
+		assertNotNull(value);
+		assertEquals(String.class, value.getClass());
+		assertEquals("99-value", value);
+	}
+
+	@Test
+	public void resolveProvidedArgFirst() {
+		this.resolvers.add(new StubArgumentResolver(1));
+		this.resolvers.add(new StubArgumentResolver("value1"));
+		Method method = ResolvableMethod.on(Handler.class).mockCall(c -> c.handle(0, "")).method();
+		Object value = invokeAndBlock(new Handler(), method, 2, "value2");
+
+		assertEquals("2-value2", value);
+	}
+
+	@Test
+	public void exceptionInResolvingArg() {
+		this.resolvers.add(new InvocableHandlerMethodTests.ExceptionRaisingArgumentResolver());
+		try {
+			Method method = ResolvableMethod.on(Handler.class).mockCall(c -> c.handle(0, "")).method();
+			invokeAndBlock(new Handler(), method);
+			fail("Expected exception");
+		}
+		catch (IllegalArgumentException ex) {
+			// expected -  allow HandlerMethodArgumentResolver exceptions to propagate
+		}
+	}
+
+	@Test
+	public void illegalArgumentException() {
+		this.resolvers.add(new StubArgumentResolver(Integer.class, "__not_an_int__"));
+		this.resolvers.add(new StubArgumentResolver("value"));
+		try {
+			Method method = ResolvableMethod.on(Handler.class).mockCall(c -> c.handle(0, "")).method();
+			invokeAndBlock(new Handler(), method);
+			fail("Expected exception");
+		}
+		catch (IllegalStateException ex) {
+			assertNotNull("Exception not wrapped", ex.getCause());
+			assertTrue(ex.getCause() instanceof IllegalArgumentException);
+			assertTrue(ex.getMessage().contains("Endpoint ["));
+			assertTrue(ex.getMessage().contains("Method ["));
+			assertTrue(ex.getMessage().contains("with argument values:"));
+			assertTrue(ex.getMessage().contains("[0] [type=java.lang.String] [value=__not_an_int__]"));
+			assertTrue(ex.getMessage().contains("[1] [type=java.lang.String] [value=value"));
+		}
+	}
+
+	@Test
+	public void invocationTargetException() {
+		Method method = ResolvableMethod.on(Handler.class).argTypes(Throwable.class).resolveMethod();
+
+		Throwable expected = new Throwable("error");
+		Mono<Object> result = invoke(new Handler(), method, expected);
+		StepVerifier.create(result).expectErrorSatisfies(actual -> assertSame(expected, actual)).verify();
+	}
+
+	@Test
+	public void voidMethod() {
+		this.resolvers.add(new StubArgumentResolver(double.class, 5.25));
+		Method method = ResolvableMethod.on(Handler.class).mockCall(c -> c.handle(0.0d)).method();
+		Handler handler = new Handler();
+		Object value = invokeAndBlock(handler, method);
+
+		assertNull(value);
+		assertEquals(1, getStubResolver(0).getResolvedParameters().size());
+		assertEquals("5.25", handler.getResult());
+		assertEquals("amount", getStubResolver(0).getResolvedParameters().get(0).getParameterName());
+	}
+
+	@Test
+	public void voidMonoMethod() {
+		Method method = ResolvableMethod.on(Handler.class).mockCall(Handler::handleAsync).method();
+		Handler handler = new Handler();
+		Object value = invokeAndBlock(handler, method);
+
+		assertNull(value);
+		assertEquals("success", handler.getResult());
+	}
+
+
+	@Nullable
+	private Object invokeAndBlock(Object handler, Method method, Object... providedArgs) {
+		return invoke(handler, method, providedArgs).block(Duration.ofSeconds(5));
+	}
+
+	private Mono<Object> invoke(Object handler, Method method, Object... providedArgs) {
+		InvocableHandlerMethod handlerMethod = new InvocableHandlerMethod(handler, method);
+		handlerMethod.setArgumentResolvers(this.resolvers);
+		return handlerMethod.invoke(this.message, providedArgs);
+	}
+
+	private StubArgumentResolver getStubResolver(int index) {
+		return (StubArgumentResolver) this.resolvers.get(index);
+	}
+
+
+
+	@SuppressWarnings({"unused", "UnusedReturnValue", "SameParameterValue"})
+	private static class Handler {
+
+		private AtomicReference<String> result = new AtomicReference<>();
+
+
+		public String getResult() {
+			return this.result.get();
+		}
+
+		String handle(Integer intArg, String stringArg) {
+			return intArg + "-" + stringArg;
+		}
+
+		void handle(double amount) {
+			this.result.set(String.valueOf(amount));
+		}
+
+		void handleWithException(Throwable ex) throws Throwable {
+			throw ex;
+		}
+
+		Mono<Void> handleAsync() {
+			return Mono.delay(Duration.ofMillis(100)).thenEmpty(Mono.defer(() -> {
+				this.result.set("success");
+				return Mono.empty();
+			}));
+		}
+	}
+
+
+	private static class ExceptionRaisingArgumentResolver implements HandlerMethodArgumentResolver {
+
+		@Override
+		public boolean supportsParameter(MethodParameter parameter) {
+			return true;
+		}
+
+		@Override
+		public Mono<Object> resolveArgument(MethodParameter parameter, Message<?> message) {
+			return Mono.error(new IllegalArgumentException("oops, can't read"));
+		}
+	}
+
+}

spring-messaging/src/test/java/org/springframework/messaging/handler/invocation/reactive/StubArgumentResolver.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2002-2018 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 java.util.ArrayList;
+import java.util.List;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+
+/**
+ * Stub resolver for a fixed value type and/or value.
+ *
+ * @author Rossen Stoyanchev
+ */
+public class StubArgumentResolver implements HandlerMethodArgumentResolver {
+
+	private final Class<?> valueType;
+
+	@Nullable
+	private final Object value;
+
+	private List<MethodParameter> resolvedParameters = new ArrayList<>();
+
+
+	public StubArgumentResolver(Object value) {
+		this(value.getClass(), value);
+	}
+
+	public StubArgumentResolver(Class<?> valueType) {
+		this(valueType, null);
+	}
+
+	public StubArgumentResolver(Class<?> valueType, Object value) {
+		this.valueType = valueType;
+		this.value = value;
+	}
+
+
+	public List<MethodParameter> getResolvedParameters() {
+		return resolvedParameters;
+	}
+
+
+	@Override
+	public boolean supportsParameter(MethodParameter parameter) {
+		return parameter.getParameterType().equals(this.valueType);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public Mono<Object> resolveArgument(MethodParameter parameter, Message<?> message) {
+		this.resolvedParameters.add(parameter);
+		return Mono.justOrEmpty(this.value);
+	}
+
+}