Add support for suspending handler methods in WebFlux

This commit turns Coroutines suspending methods to `Mono` which can be
handled natively by WebFlux.

See gh-19975

spring-webflux/src/main/java/org/springframework/web/reactive/result/method/InvocableHandlerMethod.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * 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.
@@ -16,6 +16,8 @@
 
 package org.springframework.web.reactive.result.method;
 
+import static org.springframework.web.reactive.result.method.InvocableHandlerMethodKt.*;
+
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
@@ -27,6 +29,7 @@ import java.util.stream.Stream;
 import reactor.core.publisher.Mono;
 
 import org.springframework.core.DefaultParameterNameDiscoverer;
+import org.springframework.core.KotlinDetector;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ParameterNameDiscoverer;
 import org.springframework.core.ReactiveAdapter;
@@ -48,6 +51,7 @@ import org.springframework.web.server.ServerWebExchange;
  *
  * @author Rossen Stoyanchev
  * @author Juergen Hoeller
+ * @author Sebastien Deleuze
  * @since 5.0
  */
 public class InvocableHandlerMethod extends HandlerMethod {
@@ -136,7 +140,13 @@ public class InvocableHandlerMethod extends HandlerMethod {
 			Object value;
 			try {
 				ReflectionUtils.makeAccessible(getBridgedMethod());
-				value = getBridgedMethod().invoke(getBean(), args);
+				Method method = getBridgedMethod();
+				if (KotlinDetector.isKotlinReflectPresent() && KotlinDetector.isKotlinType(method.getDeclaringClass())) {
+					value = invokeHandlerMethod(method, getBean(), args);
+				}
+				else {
+					value = method.invoke(getBean(), args);
+				}
 			}
 			catch (IllegalArgumentException ex) {
 				assertTargetBean(getBridgedMethod(), getBean(), args);

spring-webflux/src/main/java/org/springframework/web/reactive/result/method/annotation/ContinuationHandlerMethodArgumentResolver.java

@@ -0,0 +1,43 @@
+/*
+ * 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.web.reactive.result.method.annotation;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.web.reactive.BindingContext;
+import org.springframework.web.reactive.result.method.HandlerMethodArgumentResolver;
+import org.springframework.web.server.ServerWebExchange;
+
+/**
+ * No-op resolver for method arguments of type {@link kotlin.coroutines.Continuation}.
+ *
+ * @author Sebastien Deleuze
+ * @since 5.2
+ */
+public class ContinuationHandlerMethodArgumentResolver implements HandlerMethodArgumentResolver {
+
+	@Override
+	public boolean supportsParameter(MethodParameter parameter) {
+		return "kotlin.coroutines.Continuation".equals(parameter.getParameterType().getName());
+	}
+
+	@Override
+	public Mono<Object> resolveArgument(MethodParameter parameter, BindingContext bindingContext, ServerWebExchange exchange) {
+		return Mono.empty();
+	}
+}

spring-webflux/src/main/java/org/springframework/web/reactive/result/method/annotation/ControllerMethodResolver.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * 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.
@@ -32,6 +32,7 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.core.KotlinDetector;
 import org.springframework.core.MethodIntrospector;
 import org.springframework.core.ReactiveAdapterRegistry;
 import org.springframework.core.annotation.AnnotatedElementUtils;
@@ -190,6 +191,9 @@ class ControllerMethodResolver {
 		result.add(new RequestAttributeMethodArgumentResolver(beanFactory, reactiveRegistry));
 
 		// Type-based...
+		if (KotlinDetector.isKotlinPresent()) {
+			result.add(new ContinuationHandlerMethodArgumentResolver());
+		}
 		if (!readers.isEmpty()) {
 			result.add(new HttpEntityArgumentResolver(readers, reactiveRegistry));
 		}

spring-webflux/src/main/kotlin/org/springframework/web/reactive/result/method/InvocableHandlerMethod.kt

@@ -0,0 +1,43 @@
+/*
+ * 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.web.reactive.result.method
+
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.reactor.mono
+import reactor.core.publisher.onErrorMap
+import java.lang.reflect.InvocationTargetException
+import java.lang.reflect.Method
+import kotlin.reflect.full.callSuspend
+import kotlin.reflect.jvm.kotlinFunction
+
+/**
+ * Invoke an handler method converting suspending method to {@link Mono} if necessary.
+ *
+ * @author Sebastien Deleuze
+ * @since 5.2
+ */
+internal fun invokeHandlerMethod(method: Method, bean: Any, vararg args: Any?): Any? {
+	val function = method.kotlinFunction!!
+	return if (function.isSuspend) {
+		GlobalScope.mono { function.callSuspend(bean, *args.sliceArray(0..(args.size-2)))
+				.let { if (it == Unit) null else it} }
+				.onErrorMap(InvocationTargetException::class) { it.targetException }
+	}
+	else {
+		function.call(bean, *args)
+	}
+}

spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/ControllerMethodResolverTests.java

@@ -105,6 +105,7 @@ public class ControllerMethodResolverTests {
 		assertEquals(SessionAttributeMethodArgumentResolver.class, next(resolvers, index).getClass());
 		assertEquals(RequestAttributeMethodArgumentResolver.class, next(resolvers, index).getClass());
 
+		assertEquals(ContinuationHandlerMethodArgumentResolver.class, next(resolvers, index).getClass());
 		assertEquals(HttpEntityArgumentResolver.class, next(resolvers, index).getClass());
 		assertEquals(ModelArgumentResolver.class, next(resolvers, index).getClass());
 		assertEquals(ErrorsMethodArgumentResolver.class, next(resolvers, index).getClass());
@@ -143,6 +144,7 @@ public class ControllerMethodResolverTests {
 		assertEquals(SessionAttributeMethodArgumentResolver.class, next(resolvers, index).getClass());
 		assertEquals(RequestAttributeMethodArgumentResolver.class, next(resolvers, index).getClass());
 
+		assertEquals(ContinuationHandlerMethodArgumentResolver.class, next(resolvers, index).getClass());
 		assertEquals(ModelArgumentResolver.class, next(resolvers, index).getClass());
 		assertEquals(ErrorsMethodArgumentResolver.class, next(resolvers, index).getClass());
 		assertEquals(ServerWebExchangeArgumentResolver.class, next(resolvers, index).getClass());
@@ -209,6 +211,7 @@ public class ControllerMethodResolverTests {
 		assertEquals(SessionAttributeMethodArgumentResolver.class, next(resolvers, index).getClass());
 		assertEquals(RequestAttributeMethodArgumentResolver.class, next(resolvers, index).getClass());
 
+		assertEquals(ContinuationHandlerMethodArgumentResolver.class, next(resolvers, index).getClass());
 		assertEquals(ModelArgumentResolver.class, next(resolvers, index).getClass());
 		assertEquals(ServerWebExchangeArgumentResolver.class, next(resolvers, index).getClass());
 		assertEquals(PrincipalArgumentResolver.class, next(resolvers, index).getClass());

spring-webflux/src/test/kotlin/org/springframework/web/reactive/result/KotlinInvocableHandlerMethodTests.kt

@@ -0,0 +1,155 @@
+/*
+ * 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.web.reactive.result
+
+import io.mockk.every
+import io.mockk.mockk
+import kotlinx.coroutines.delay
+import org.hamcrest.CoreMatchers.`is`
+import org.hamcrest.MatcherAssert.assertThat
+import org.junit.Assert.assertEquals
+import org.junit.Test
+import org.springframework.http.HttpStatus
+import org.springframework.http.server.reactive.ServerHttpResponse
+import org.springframework.mock.http.server.reactive.test.MockServerHttpRequest.get
+import org.springframework.mock.web.test.server.MockServerWebExchange
+import org.springframework.web.bind.annotation.ResponseStatus
+import org.springframework.web.reactive.BindingContext
+import org.springframework.web.reactive.HandlerResult
+import org.springframework.web.reactive.result.method.HandlerMethodArgumentResolver
+import org.springframework.web.reactive.result.method.InvocableHandlerMethod
+import org.springframework.web.reactive.result.method.annotation.ContinuationHandlerMethodArgumentResolver
+import reactor.core.publisher.Mono
+import reactor.test.StepVerifier
+import reactor.test.expectError
+import java.lang.reflect.Method
+import kotlin.reflect.jvm.javaMethod
+
+class KotlinInvocableHandlerMethodTests {
+
+	private val exchange = MockServerWebExchange.from(get("http://localhost:8080/path"))
+
+	private val resolvers = mutableListOf<HandlerMethodArgumentResolver>(ContinuationHandlerMethodArgumentResolver())
+
+	@Test
+	fun resolveNoArg() {
+		this.resolvers.add(stubResolver(Mono.empty()))
+		val method = CoroutinesController::singleArg.javaMethod!!
+		val result = invoke(CoroutinesController(), method, null)
+		assertHandlerResultValue(result, "success:null")
+	}
+
+	@Test
+	fun resolveArg() {
+		this.resolvers.add(stubResolver("foo"))
+		val method = CoroutinesController::singleArg.javaMethod!!
+		val result = invoke(CoroutinesController(), method,"foo")
+		assertHandlerResultValue(result, "success:foo")
+	}
+
+	@Test
+	fun resolveNoArgs() {
+		val method = CoroutinesController::noArgs.javaMethod!!
+		val result = invoke(CoroutinesController(), method)
+		assertHandlerResultValue(result, "success")
+	}
+
+	@Test
+	fun invocationTargetException() {
+		val method = CoroutinesController::exceptionMethod.javaMethod!!
+		val result = invoke(CoroutinesController(), method)
+
+		StepVerifier.create(result)
+				.consumeNextWith { StepVerifier.create(it.returnValue as Mono<*>).expectError(IllegalStateException::class).verify() }
+				.verifyComplete()
+	}
+
+	@Test
+	fun responseStatusAnnotation() {
+		val method = CoroutinesController::created.javaMethod!!
+		val result = invoke(CoroutinesController(), method)
+
+		assertHandlerResultValue(result, "created")
+		assertThat<HttpStatus>(this.exchange.response.statusCode, `is`(HttpStatus.CREATED))
+	}
+
+	@Test
+	fun voidMethodWithResponseArg() {
+		val response = this.exchange.response
+		this.resolvers.add(stubResolver(response))
+		val method = CoroutinesController::response.javaMethod!!
+		val result = invoke(CoroutinesController(), method)
+
+		StepVerifier.create(result)
+				.consumeNextWith { StepVerifier.create(it.returnValue as Mono<*>).verifyComplete() }
+				.verifyComplete()
+		assertEquals("bar", this.exchange.response.headers.getFirst("foo"))
+	}
+
+	private fun invoke(handler: Any, method: Method, vararg providedArgs: Any?): Mono<HandlerResult> {
+		val invocable = InvocableHandlerMethod(handler, method)
+		invocable.setArgumentResolvers(this.resolvers)
+		return invocable.invoke(this.exchange, BindingContext(), *providedArgs)
+	}
+
+	private fun stubResolver(stubValue: Any?): HandlerMethodArgumentResolver {
+		return stubResolver(Mono.justOrEmpty(stubValue))
+	}
+
+	private fun stubResolver(stubValue: Mono<Any>): HandlerMethodArgumentResolver {
+		val resolver = mockk<HandlerMethodArgumentResolver>()
+		every { resolver.supportsParameter(any()) } returns true
+		every { resolver.resolveArgument(any(), any(), any()) } returns stubValue
+		return resolver
+	}
+
+	private fun assertHandlerResultValue(mono: Mono<HandlerResult>, expected: String) {
+		StepVerifier.create(mono)
+				.consumeNextWith { StepVerifier.create(it.returnValue as Mono<*>).expectNext(expected).verifyComplete() }
+				.verifyComplete()
+	}
+
+	class CoroutinesController {
+
+		suspend fun singleArg(q: String?): String {
+			delay(10)
+			return "success:$q"
+		}
+
+		suspend fun noArgs(): String {
+			delay(10)
+			return "success"
+		}
+
+		suspend fun exceptionMethod() {
+			throw IllegalStateException("boo")
+		}
+
+		@ResponseStatus(HttpStatus.CREATED)
+		suspend fun created(): String {
+			delay(10)
+			return "created"
+		}
+
+		suspend fun response(response: ServerHttpResponse) {
+			delay(10)
+			response.headers.add("foo", "bar")
+		}
+
+
+	}
+}